diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java new file mode 100644 index 0000000..d642f9a --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java @@ -0,0 +1,57 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.domain.execution.DeviceCommandType; + +public final class DeviceCommandEnvelope { + private final String commandId; + private final String commandKey; + private final String deviceId; + private final DeviceCommandType type; + private final String payload; + private final int attempt; + private final long topologyVersion; + private final long fencingToken; + + public DeviceCommandEnvelope(String commandId, String commandKey, + String deviceId, DeviceCommandType type, + String payload, int attempt, + long topologyVersion, long fencingToken) { + this.commandId = requireText(commandId, "commandId"); + this.commandKey = requireText(commandKey, "commandKey"); + this.deviceId = requireText(deviceId, "deviceId"); + this.type = java.util.Objects.requireNonNull(type, "type"); + this.payload = requireText(payload, "payload"); + if (attempt <= 0 || topologyVersion <= 0L || fencingToken <= 0L) { + throw new IllegalArgumentException("command envelope versions are invalid"); + } + this.attempt = attempt; + this.topologyVersion = topologyVersion; + this.fencingToken = fencingToken; + } + + public static DeviceCommandEnvelope from(DeviceCommand command, + long topologyVersion, + long fencingToken) { + return new DeviceCommandEnvelope(command.getCommandId(), + command.getCommandKey(), command.getDeviceId(), command.getType(), + command.getPayload(), command.getAttempt(), topologyVersion, + fencingToken); + } + + public String getCommandId() { return commandId; } + public String getCommandKey() { return commandKey; } + public String getDeviceId() { return deviceId; } + public DeviceCommandType getType() { return type; } + public String getPayload() { return payload; } + public int getAttempt() { return attempt; } + public long getTopologyVersion() { return topologyVersion; } + public long getFencingToken() { return fencingToken; } + + private static String requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(field + " must not be blank"); + } + return value.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/EventAcceptanceResult.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/EventAcceptanceResult.java new file mode 100644 index 0000000..6a0ad53 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/EventAcceptanceResult.java @@ -0,0 +1,8 @@ +package com.ruoyi.wcs.application.device; + +public enum EventAcceptanceResult { + APPLIED, + DUPLICATE, + STALE, + RECOVERY_REQUIRED +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java new file mode 100644 index 0000000..0bd7496 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java @@ -0,0 +1,259 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.CommandResolutionEvidence; +import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution; +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.domain.execution.DeviceCommandState; +import com.ruoyi.wcs.domain.execution.DeviceEvent; +import com.ruoyi.wcs.domain.execution.DeviceSnapshot; +import com.ruoyi.wcs.domain.shared.DomainRuleViolation; +import com.ruoyi.wcs.port.in.DeviceEventUseCase; +import com.ruoyi.wcs.port.out.DeviceCommandRepository; +import com.ruoyi.wcs.port.out.DeviceEventRepository; +import com.ruoyi.wcs.port.out.DeviceExecutionRepository; +import com.ruoyi.wcs.port.out.RecoveryCaseOpener; +import com.ruoyi.wcs.port.out.ResourceReservationRepository; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Instant; +import java.util.Collections; +import java.util.List; + +public final class FeedbackReconciler implements DeviceEventUseCase { + private final DeviceCommandRepository commandRepository; + private final DeviceExecutionRepository executionRepository; + private final DeviceEventRepository eventRepository; + private final ResourceReservationRepository reservationRepository; + private final RecoveryCaseOpener recoveryCaseOpener; + + public FeedbackReconciler(DeviceCommandRepository commandRepository, + DeviceExecutionRepository executionRepository, + DeviceEventRepository eventRepository, + ResourceReservationRepository reservationRepository, + RecoveryCaseOpener recoveryCaseOpener) { + this.commandRepository = java.util.Objects.requireNonNull( + commandRepository, "commandRepository"); + this.executionRepository = java.util.Objects.requireNonNull( + executionRepository, "executionRepository"); + this.eventRepository = java.util.Objects.requireNonNull( + eventRepository, "eventRepository"); + this.reservationRepository = java.util.Objects.requireNonNull( + reservationRepository, "reservationRepository"); + this.recoveryCaseOpener = java.util.Objects.requireNonNull( + recoveryCaseOpener, "recoveryCaseOpener"); + } + + @Override + @Transactional + public EventAcceptanceResult accept(DeviceEvent event) { + if (!eventRepository.claim(event)) { + return EventAcceptanceResult.DUPLICATE; + } + CommittedDeviceExecution execution = executionRepository + .findByCommandKey(event.getCommandKey()); + if (execution == null || !execution.getDeviceId().equals(event.getDeviceId())) { + openRecovery(event.getDeviceId(), event.getCommandKey(), + "COMMAND_CONTEXT_MISMATCH", event.getEventId(), + event.getReceivedTime()); + return EventAcceptanceResult.RECOVERY_REQUIRED; + } + if (event.getDeviceSequence() <= execution.getLastDeviceSequence()) { + return EventAcceptanceResult.STALE; + } + if (!execution.getTaskNumber().equals(event.getTaskNumber())) { + openRecovery(execution.getDeviceId(), execution.getCommandKey(), + "TASK_NUMBER_MISMATCH", event.getEventId(), + event.getReceivedTime()); + return EventAcceptanceResult.RECOVERY_REQUIRED; + } + + DeviceCommand command = requireCommand(execution.getCommandKey()); + try { + switch (event.getEventType()) { + case ACKNOWLEDGED: + requireActive(command, false); + command.acknowledge(event.getEventId(), + transitionTime(command, event.getReceivedTime())); + executionRepository.save(execution.observeSequence( + event.getDeviceSequence())); + break; + case PROGRESS: + requireActive(command, false); + applyProgress(execution, event.getDeviceSequence(), + event.getCurrentNodeId(), event.getRemainingRouteNodes(), + false, event.getEventId()); + break; + case COMPLETED: + requireActive(command, true); + applyProgress(execution, event.getDeviceSequence(), + event.getCurrentNodeId(), event.getRemainingRouteNodes(), + true, event.getEventId()); + complete(command, event.getEventId(), event.getReceivedTime()); + break; + case REJECTED: + requireActive(command, false); + command.reject("DEVICE_REJECTED:" + event.getEventId(), + transitionTime(command, event.getReceivedTime())); + executionRepository.save(execution.observeSequence( + event.getDeviceSequence())); + break; + case FAULT: + openRecovery(execution.getDeviceId(), execution.getCommandKey(), + "DEVICE_FAULT", event.getEventId(), + event.getReceivedTime()); + return EventAcceptanceResult.RECOVERY_REQUIRED; + default: + throw new IllegalStateException("unsupported device event"); + } + } catch (DomainRuleViolation ex) { + openRecovery(execution.getDeviceId(), execution.getCommandKey(), + reason(ex), event.getEventId(), event.getReceivedTime()); + return EventAcceptanceResult.RECOVERY_REQUIRED; + } + commandRepository.save(command); + return EventAcceptanceResult.APPLIED; + } + + @Transactional + public ReconcileResult reconcile(DeviceSnapshot snapshot) { + CommittedDeviceExecution execution = executionRepository + .findActiveByDeviceId(snapshot.getDeviceId()); + if (execution == null) { + return recovery(snapshot, null, "ACTIVE_EXECUTION_NOT_FOUND"); + } + if (!snapshot.isConnected()) { + return recovery(snapshot, execution, "COMMUNICATION_LOST"); + } + if (!execution.getCommandKey().equals(snapshot.getCommandKey())) { + return recovery(snapshot, execution, "COMMAND_KEY_MISMATCH"); + } + if (!execution.getTaskNumber().equals(snapshot.getTaskNumber())) { + return recovery(snapshot, execution, "TASK_NUMBER_MISMATCH"); + } + if (!execution.getExpectedBootId().equals(snapshot.getDeviceBootId())) { + return recovery(snapshot, execution, "DEVICE_RESTARTED"); + } + if (execution.getTopologyVersion() != snapshot.getTopologyVersion()) { + return recovery(snapshot, execution, "TOPOLOGY_VERSION_MISMATCH"); + } + if (execution.getFencingToken() != snapshot.getFencingToken()) { + return recovery(snapshot, execution, "FENCING_TOKEN_MISMATCH"); + } + if (snapshot.getDeviceSequence() <= execution.getLastDeviceSequence()) { + return new ReconcileResult(ReconcileStatus.IGNORED_STALE, + execution.getCommandKey(), Collections.emptyList(), null); + } + + DeviceCommand command = requireCommand(execution.getCommandKey()); + try { + if (command.getState() == DeviceCommandState.UNKNOWN + && !snapshot.isCommandCompleted()) { + execution.evaluateProgress(snapshot.getDeviceSequence(), + snapshot.getCurrentNodeId(), + snapshot.getRemainingRouteNodes(), false); + return new ReconcileResult(ReconcileStatus.CONSISTENT, + execution.getCommandKey(), + Collections.emptyList(), + "UNKNOWN_DIAGNOSTIC_ONLY"); + } + requireActive(command, snapshot.isCommandCompleted()); + List released = applyProgress(execution, + snapshot.getDeviceSequence(), snapshot.getCurrentNodeId(), + snapshot.getRemainingRouteNodes(), snapshot.isCommandCompleted(), + snapshot.getSnapshotId()); + if (snapshot.isCommandCompleted()) { + complete(command, snapshot.getSnapshotId(), snapshot.getObservedAt()); + commandRepository.save(command); + } + return new ReconcileResult(released.isEmpty() + ? ReconcileStatus.CONSISTENT : ReconcileStatus.ADVANCED, + execution.getCommandKey(), released, null); + } catch (DomainRuleViolation ex) { + return recovery(snapshot, execution, reason(ex)); + } + } + + private List applyProgress(CommittedDeviceExecution execution, + long sequence, String currentNode, + List remaining, boolean completed, + String evidenceId) { + CommittedDeviceExecution.Progress progress = execution.evaluateProgress( + sequence, currentNode, remaining, completed); + if (progress.isStale()) { + return Collections.emptyList(); + } + List released = progress.getReleasedResourceKeys(); + if (!released.isEmpty()) { + reservationRepository.releasePassedResources( + execution.getCommandKey(), released, evidenceId, + execution.getFencingToken()); + } + executionRepository.save(execution.advance(progress)); + return released; + } + + private static void requireActive(DeviceCommand command, + boolean physicalCompletion) { + DeviceCommandState state = command.getState(); + boolean allowed = state == DeviceCommandState.SENT + || state == DeviceCommandState.ACKNOWLEDGED + || (physicalCompletion && state == DeviceCommandState.UNKNOWN); + if (!allowed) { + throw new DomainRuleViolation("COMMAND_STATE_MISMATCH"); + } + } + + private void complete(DeviceCommand command, String evidenceId, Instant at) { + Instant transitionAt = transitionTime(command, at); + if (command.getState() == DeviceCommandState.UNKNOWN) { + command.succeedAfterRecovery( + CommandResolutionEvidence.physicalCompletion( + command.getCommandKey(), evidenceId), transitionAt); + } else { + command.succeed(evidenceId, transitionAt); + } + } + + private ReconcileResult recovery(DeviceSnapshot snapshot, + CommittedDeviceExecution execution, + String reason) { + String commandKey = execution == null + ? snapshot.getCommandKey() : execution.getCommandKey(); + openRecovery(snapshot.getDeviceId(), commandKey, reason, + snapshot.getSnapshotId(), snapshot.getObservedAt()); + return new ReconcileResult(ReconcileStatus.RECOVERY_REQUIRED, + commandKey, Collections.emptyList(), reason); + } + + private void openRecovery(String deviceId, String commandKey, String reason, + String evidenceId, Instant evidenceTime) { + DeviceCommand command = commandRepository.findByCommandKey(commandKey); + if (command != null && (command.getState() == DeviceCommandState.SENT + || command.getState() == DeviceCommandState.ACKNOWLEDGED)) { + command.markUnknown(reason, transitionTime(command, evidenceTime)); + commandRepository.save(command); + } + reservationRepository.protectForUnknown(commandKey, evidenceId); + recoveryCaseOpener.open(deviceId, commandKey, reason, + Collections.singletonList(evidenceId)); + } + + private DeviceCommand requireCommand(String commandKey) { + DeviceCommand command = commandRepository.findByCommandKey(commandKey); + if (command == null) { + throw new DomainRuleViolation("COMMAND_NOT_FOUND"); + } + return command; + } + + private static Instant transitionTime(DeviceCommand command, Instant candidate) { + return candidate.isBefore(command.getUpdatedAt()) + ? command.getUpdatedAt() : candidate; + } + + private static String reason(DomainRuleViolation violation) { + String message = violation.getMessage(); + return message == null || message.trim().isEmpty() + ? "RECONCILIATION_FAILED" : message; + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/ReconcileResult.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/ReconcileResult.java new file mode 100644 index 0000000..85458d5 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/ReconcileResult.java @@ -0,0 +1,29 @@ +package com.ruoyi.wcs.application.device; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public final class ReconcileResult { + private final ReconcileStatus status; + private final String commandKey; + private final List releasedResourceKeys; + private final String recoveryReason; + + public ReconcileResult(ReconcileStatus status, String commandKey, + List releasedResourceKeys, + String recoveryReason) { + this.status = java.util.Objects.requireNonNull(status, "status"); + this.commandKey = commandKey; + this.releasedResourceKeys = Collections.unmodifiableList( + new ArrayList(releasedResourceKeys)); + this.recoveryReason = recoveryReason; + } + + public ReconcileStatus getStatus() { return status; } + public String getCommandKey() { return commandKey; } + public List getReleasedResourceKeys() { + return releasedResourceKeys; + } + public String getRecoveryReason() { return recoveryReason; } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/ReconcileStatus.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/ReconcileStatus.java new file mode 100644 index 0000000..f45ea03 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/ReconcileStatus.java @@ -0,0 +1,8 @@ +package com.ruoyi.wcs.application.device; + +public enum ReconcileStatus { + CONSISTENT, + ADVANCED, + IGNORED_STALE, + RECOVERY_REQUIRED +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java new file mode 100644 index 0000000..925a662 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java @@ -0,0 +1,49 @@ +package com.ruoyi.wcs.application.device; + +public final class SendReceipt { + public enum Status { + ACCEPTED, + ACCEPTED_UNKNOWN, + NOT_SENT, + REJECTED + } + + private final Status status; + private final String receiptId; + private final String detail; + + public SendReceipt(Status status, String receiptId, String detail) { + this.status = java.util.Objects.requireNonNull(status, "status"); + this.receiptId = receiptId; + this.detail = detail; + if ((status == Status.ACCEPTED || status == Status.ACCEPTED_UNKNOWN) + && isBlank(receiptId)) { + throw new IllegalArgumentException( + "accepted send receipt must have a receiptId"); + } + } + + public static SendReceipt accepted(String receiptId) { + return new SendReceipt(Status.ACCEPTED, receiptId, null); + } + + public static SendReceipt acceptedUnknown(String receiptId, String detail) { + return new SendReceipt(Status.ACCEPTED_UNKNOWN, receiptId, detail); + } + + public static SendReceipt notSent(String detail) { + return new SendReceipt(Status.NOT_SENT, null, detail); + } + + public static SendReceipt rejected(String detail) { + return new SendReceipt(Status.REJECTED, null, detail); + } + + public Status getStatus() { return status; } + public String getReceiptId() { return receiptId; } + public String getDetail() { return detail; } + + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommittedDeviceExecution.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommittedDeviceExecution.java new file mode 100644 index 0000000..82fdbe5 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommittedDeviceExecution.java @@ -0,0 +1,212 @@ +package com.ruoyi.wcs.domain.execution; + +import com.ruoyi.wcs.domain.shared.DomainRuleViolation; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public final class CommittedDeviceExecution { + private final String commandId; + private final String commandKey; + private final String deviceId; + private final String taskNumber; + private final List routeNodeIds; + private final List routeResourceKeys; + private final long topologyVersion; + private final long fencingToken; + private final String expectedBootId; + private final int currentNodeIndex; + private final int releasedResourceCount; + private final long lastDeviceSequence; + + public CommittedDeviceExecution(String commandId, String commandKey, + String deviceId, String taskNumber, + List routeNodeIds, + List routeResourceKeys, + long topologyVersion, long fencingToken, + String expectedBootId) { + this(commandId, commandKey, deviceId, taskNumber, routeNodeIds, + routeResourceKeys, topologyVersion, fencingToken, expectedBootId, + 0, 0, -1L); + } + + private CommittedDeviceExecution(String commandId, String commandKey, + String deviceId, String taskNumber, + List routeNodeIds, + List routeResourceKeys, + long topologyVersion, long fencingToken, + String expectedBootId, int currentNodeIndex, + int releasedResourceCount, + long lastDeviceSequence) { + this.commandId = requireText(commandId, "commandId"); + this.commandKey = requireText(commandKey, "commandKey"); + this.deviceId = requireText(deviceId, "deviceId"); + this.taskNumber = requireText(taskNumber, "taskNumber"); + this.routeNodeIds = immutable(routeNodeIds, "routeNodeIds"); + this.routeResourceKeys = immutable(routeResourceKeys, + "routeResourceKeys"); + if (this.routeResourceKeys.size() != this.routeNodeIds.size() * 2 - 1) { + throw new IllegalArgumentException( + "route resources must alternate node and edge resources"); + } + if (topologyVersion <= 0L || fencingToken <= 0L) { + throw new IllegalArgumentException("execution versions are invalid"); + } + this.topologyVersion = topologyVersion; + this.fencingToken = fencingToken; + this.expectedBootId = requireText(expectedBootId, "expectedBootId"); + this.currentNodeIndex = currentNodeIndex; + this.releasedResourceCount = releasedResourceCount; + this.lastDeviceSequence = lastDeviceSequence; + } + + public Progress evaluateProgress(long deviceSequence, String currentNodeId, + List remainingRouteNodes, + boolean completed) { + if (deviceSequence <= lastDeviceSequence) { + return Progress.stale(commandKey, currentNodeIndex, + releasedResourceCount, lastDeviceSequence); + } + String current = requireText(currentNodeId, "currentNodeId"); + int index = routeNodeIds.indexOf(current); + if (index < 0) { + throw new DomainRuleViolation("PATH_MISMATCH"); + } + if (index < currentNodeIndex) { + throw new DomainRuleViolation("ROUTE_REGRESSION"); + } + List remaining = immutable(remainingRouteNodes, + "remainingRouteNodes"); + if (!routeNodeIds.subList(index, routeNodeIds.size()).equals(remaining)) { + throw new DomainRuleViolation("PATH_MISMATCH"); + } + if (completed && index != routeNodeIds.size() - 1) { + throw new DomainRuleViolation("COMPLETION_SITE_MISMATCH"); + } + + int releaseUntil = completed ? routeResourceKeys.size() : index * 2; + List released = releaseUntil <= releasedResourceCount + ? Collections.emptyList() + : new ArrayList(routeResourceKeys.subList( + releasedResourceCount, releaseUntil)); + boolean advanced = index > currentNodeIndex || !released.isEmpty(); + return Progress.accepted(commandKey, currentNodeIndex, + releasedResourceCount, lastDeviceSequence, index, + Math.max(releasedResourceCount, releaseUntil), deviceSequence, + released, advanced); + } + + public CommittedDeviceExecution advance(Progress progress) { + if (progress == null || !commandKey.equals(progress.commandKey)) { + throw new DomainRuleViolation("progress belongs to another command"); + } + if (progress.stale) { + return this; + } + if (currentNodeIndex != progress.baseNodeIndex + || releasedResourceCount != progress.baseReleasedResourceCount + || lastDeviceSequence != progress.baseDeviceSequence) { + throw new DomainRuleViolation("execution progress version conflict"); + } + return new CommittedDeviceExecution(commandId, commandKey, deviceId, + taskNumber, routeNodeIds, routeResourceKeys, topologyVersion, + fencingToken, expectedBootId, progress.nextNodeIndex, + progress.nextReleasedResourceCount, progress.nextDeviceSequence); + } + + public CommittedDeviceExecution observeSequence(long deviceSequence) { + if (deviceSequence <= lastDeviceSequence) { + return this; + } + return new CommittedDeviceExecution(commandId, commandKey, deviceId, + taskNumber, routeNodeIds, routeResourceKeys, topologyVersion, + fencingToken, expectedBootId, currentNodeIndex, + releasedResourceCount, deviceSequence); + } + + public String getCommandId() { return commandId; } + public String getCommandKey() { return commandKey; } + public String getDeviceId() { return deviceId; } + public String getTaskNumber() { return taskNumber; } + public List getRouteNodeIds() { return routeNodeIds; } + public List getRouteResourceKeys() { return routeResourceKeys; } + public long getTopologyVersion() { return topologyVersion; } + public long getFencingToken() { return fencingToken; } + public String getExpectedBootId() { return expectedBootId; } + public long getLastDeviceSequence() { return lastDeviceSequence; } + + public static final class Progress { + private final String commandKey; + private final boolean stale; + private final boolean advanced; + private final int baseNodeIndex; + private final int baseReleasedResourceCount; + private final long baseDeviceSequence; + private final int nextNodeIndex; + private final int nextReleasedResourceCount; + private final long nextDeviceSequence; + private final List releasedResourceKeys; + + private Progress(String commandKey, boolean stale, boolean advanced, + int baseNodeIndex, int baseReleasedResourceCount, + long baseDeviceSequence, int nextNodeIndex, + int nextReleasedResourceCount, long nextDeviceSequence, + List releasedResourceKeys) { + this.commandKey = commandKey; + this.stale = stale; + this.advanced = advanced; + this.baseNodeIndex = baseNodeIndex; + this.baseReleasedResourceCount = baseReleasedResourceCount; + this.baseDeviceSequence = baseDeviceSequence; + this.nextNodeIndex = nextNodeIndex; + this.nextReleasedResourceCount = nextReleasedResourceCount; + this.nextDeviceSequence = nextDeviceSequence; + this.releasedResourceKeys = Collections.unmodifiableList( + new ArrayList(releasedResourceKeys)); + } + + static Progress stale(String commandKey, int nodeIndex, + int releasedResourceCount, long sequence) { + return new Progress(commandKey, true, false, nodeIndex, + releasedResourceCount, sequence, nodeIndex, + releasedResourceCount, sequence, + Collections.emptyList()); + } + + static Progress accepted(String commandKey, int baseNodeIndex, + int baseReleasedResourceCount, + long baseDeviceSequence, int nextNodeIndex, + int nextReleasedResourceCount, + long nextDeviceSequence, List released, + boolean advanced) { + return new Progress(commandKey, false, advanced, baseNodeIndex, + baseReleasedResourceCount, baseDeviceSequence, nextNodeIndex, + nextReleasedResourceCount, nextDeviceSequence, released); + } + + public boolean isStale() { return stale; } + public boolean isAdvanced() { return advanced; } + public List getReleasedResourceKeys() { + return releasedResourceKeys; + } + } + + private static List immutable(List source, String field) { + if (source == null || source.isEmpty()) { + throw new IllegalArgumentException(field + " must not be empty"); + } + List copy = new ArrayList(); + for (String value : source) { + copy.add(requireText(value, field)); + } + return Collections.unmodifiableList(copy); + } + + private static String requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(field + " must not be blank"); + } + return value.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/DeviceEvent.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/DeviceEvent.java new file mode 100644 index 0000000..eba28d7 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/DeviceEvent.java @@ -0,0 +1,116 @@ +package com.ruoyi.wcs.domain.execution; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +public final class DeviceEvent { + public enum Type { + ACKNOWLEDGED, + PROGRESS, + COMPLETED, + REJECTED, + FAULT + } + + private final String eventId; + private final String deviceId; + private final String commandKey; + private final long deviceSequence; + private final Instant deviceTime; + private final Instant receivedTime; + private final Type eventType; + private final String taskNumber; + private final String currentNodeId; + private final List remainingRouteNodes; + + public DeviceEvent(String eventId, String deviceId, String commandKey, + long deviceSequence, Instant deviceTime, + Instant receivedTime, Type eventType, String taskNumber, + String currentNodeId, List remainingRouteNodes) { + this.eventId = requireText(eventId, "eventId"); + this.deviceId = requireText(deviceId, "deviceId"); + this.commandKey = requireText(commandKey, "commandKey"); + if (deviceSequence < 0L) { + throw new IllegalArgumentException("deviceSequence must not be negative"); + } + this.deviceSequence = deviceSequence; + this.deviceTime = Objects.requireNonNull(deviceTime, "deviceTime"); + this.receivedTime = Objects.requireNonNull(receivedTime, "receivedTime"); + this.eventType = Objects.requireNonNull(eventType, "eventType"); + this.taskNumber = requireText(taskNumber, "taskNumber"); + boolean routeFactsRequired = eventType == Type.PROGRESS + || eventType == Type.COMPLETED; + if (routeFactsRequired) { + this.currentNodeId = requireText(currentNodeId, "currentNodeId"); + this.remainingRouteNodes = immutable(remainingRouteNodes, + "remainingRouteNodes"); + } else { + this.currentNodeId = currentNodeId == null ? null + : requireText(currentNodeId, "currentNodeId"); + this.remainingRouteNodes = remainingRouteNodes == null + ? Collections.emptyList() + : immutable(remainingRouteNodes, "remainingRouteNodes"); + } + } + + public String getEventId() { + return eventId; + } + + public String getDeviceId() { + return deviceId; + } + + public String getCommandKey() { + return commandKey; + } + + public long getDeviceSequence() { + return deviceSequence; + } + + public Instant getDeviceTime() { + return deviceTime; + } + + public Instant getReceivedTime() { + return receivedTime; + } + + public Type getEventType() { + return eventType; + } + + public String getTaskNumber() { + return taskNumber; + } + + public String getCurrentNodeId() { + return currentNodeId; + } + + public List getRemainingRouteNodes() { + return remainingRouteNodes; + } + + private static List immutable(List source, String field) { + if (source == null || source.isEmpty()) { + throw new IllegalArgumentException(field + " must not be empty"); + } + List copy = new ArrayList(); + for (String value : source) { + copy.add(requireText(value, field)); + } + return Collections.unmodifiableList(copy); + } + + private static String requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(field + " must not be blank"); + } + return value.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/DeviceSnapshot.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/DeviceSnapshot.java new file mode 100644 index 0000000..b359ece --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/DeviceSnapshot.java @@ -0,0 +1,80 @@ +package com.ruoyi.wcs.domain.execution; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +public final class DeviceSnapshot { + private final String snapshotId; + private final String deviceId; + private final String commandKey; + private final String taskNumber; + private final long deviceSequence; + private final Instant observedAt; + private final boolean connected; + private final String deviceBootId; + private final long topologyVersion; + private final long fencingToken; + private final String currentNodeId; + private final List remainingRouteNodes; + private final boolean commandCompleted; + + public DeviceSnapshot(String snapshotId, String deviceId, String commandKey, + String taskNumber, long deviceSequence, + Instant observedAt, boolean connected, + String deviceBootId, long topologyVersion, + long fencingToken, String currentNodeId, + List remainingRouteNodes, + boolean commandCompleted) { + this.snapshotId = requireText(snapshotId, "snapshotId"); + this.deviceId = requireText(deviceId, "deviceId"); + this.commandKey = requireText(commandKey, "commandKey"); + this.taskNumber = requireText(taskNumber, "taskNumber"); + if (deviceSequence < 0L || topologyVersion <= 0L || fencingToken <= 0L) { + throw new IllegalArgumentException("snapshot versions are invalid"); + } + this.deviceSequence = deviceSequence; + this.observedAt = Objects.requireNonNull(observedAt, "observedAt"); + this.connected = connected; + this.deviceBootId = requireText(deviceBootId, "deviceBootId"); + this.topologyVersion = topologyVersion; + this.fencingToken = fencingToken; + this.currentNodeId = requireText(currentNodeId, "currentNodeId"); + this.remainingRouteNodes = immutable(remainingRouteNodes); + this.commandCompleted = commandCompleted; + } + + public String getSnapshotId() { return snapshotId; } + public String getDeviceId() { return deviceId; } + public String getCommandKey() { return commandKey; } + public String getTaskNumber() { return taskNumber; } + public long getDeviceSequence() { return deviceSequence; } + public Instant getObservedAt() { return observedAt; } + public boolean isConnected() { return connected; } + public String getDeviceBootId() { return deviceBootId; } + public long getTopologyVersion() { return topologyVersion; } + public long getFencingToken() { return fencingToken; } + public String getCurrentNodeId() { return currentNodeId; } + public List getRemainingRouteNodes() { return remainingRouteNodes; } + public boolean isCommandCompleted() { return commandCompleted; } + + private static List immutable(List source) { + if (source == null || source.isEmpty()) { + throw new IllegalArgumentException("remainingRouteNodes must not be empty"); + } + List copy = new ArrayList(); + for (String value : source) { + copy.add(requireText(value, "remainingRouteNodes")); + } + return Collections.unmodifiableList(copy); + } + + private static String requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(field + " must not be blank"); + } + return value.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/in/DeviceEventUseCase.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/in/DeviceEventUseCase.java new file mode 100644 index 0000000..6102b66 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/in/DeviceEventUseCase.java @@ -0,0 +1,8 @@ +package com.ruoyi.wcs.port.in; + +import com.ruoyi.wcs.application.device.EventAcceptanceResult; +import com.ruoyi.wcs.domain.execution.DeviceEvent; + +public interface DeviceEventUseCase { + EventAcceptanceResult accept(DeviceEvent event); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceCommandGateway.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceCommandGateway.java new file mode 100644 index 0000000..99e2cb5 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceCommandGateway.java @@ -0,0 +1,8 @@ +package com.ruoyi.wcs.port.out; + +import com.ruoyi.wcs.application.device.DeviceCommandEnvelope; +import com.ruoyi.wcs.application.device.SendReceipt; + +public interface DeviceCommandGateway { + SendReceipt send(DeviceCommandEnvelope envelope); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceEventRepository.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceEventRepository.java new file mode 100644 index 0000000..8b86372 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceEventRepository.java @@ -0,0 +1,7 @@ +package com.ruoyi.wcs.port.out; + +import com.ruoyi.wcs.domain.execution.DeviceEvent; + +public interface DeviceEventRepository { + boolean claim(DeviceEvent event); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceExecutionRepository.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceExecutionRepository.java new file mode 100644 index 0000000..84f49eb --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceExecutionRepository.java @@ -0,0 +1,11 @@ +package com.ruoyi.wcs.port.out; + +import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution; + +public interface DeviceExecutionRepository { + CommittedDeviceExecution findByCommandKey(String commandKey); + + CommittedDeviceExecution findActiveByDeviceId(String deviceId); + + void save(CommittedDeviceExecution execution); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceSnapshotGateway.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceSnapshotGateway.java new file mode 100644 index 0000000..b8b467b --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/DeviceSnapshotGateway.java @@ -0,0 +1,7 @@ +package com.ruoyi.wcs.port.out; + +import com.ruoyi.wcs.domain.execution.DeviceSnapshot; + +public interface DeviceSnapshotGateway { + DeviceSnapshot readSnapshot(String deviceId); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/RecoveryCaseOpener.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/RecoveryCaseOpener.java new file mode 100644 index 0000000..7787faa --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/RecoveryCaseOpener.java @@ -0,0 +1,8 @@ +package com.ruoyi.wcs.port.out; + +import java.util.List; + +public interface RecoveryCaseOpener { + void open(String deviceId, String commandKey, String reason, + List evidenceIds); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/ResourceReservationRepository.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/ResourceReservationRepository.java new file mode 100644 index 0000000..e82dadc --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/ResourceReservationRepository.java @@ -0,0 +1,10 @@ +package com.ruoyi.wcs.port.out; + +import java.util.List; + +public interface ResourceReservationRepository { + void releasePassedResources(String commandKey, List resourceKeys, + String releaseEvidenceId, long fencingToken); + + void protectForUnknown(String commandKey, String evidenceId); +} diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java new file mode 100644 index 0000000..c386f77 --- /dev/null +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java @@ -0,0 +1,293 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution; +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.domain.execution.DeviceCommandState; +import com.ruoyi.wcs.domain.execution.DeviceCommandType; +import com.ruoyi.wcs.domain.execution.DeviceEvent; +import com.ruoyi.wcs.domain.execution.DeviceSnapshot; +import com.ruoyi.wcs.port.out.DeviceCommandRepository; +import com.ruoyi.wcs.port.out.DeviceEventRepository; +import com.ruoyi.wcs.port.out.DeviceExecutionRepository; +import com.ruoyi.wcs.port.out.RecoveryCaseOpener; +import com.ruoyi.wcs.port.out.ResourceReservationRepository; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class FeedbackReconcilerTest { + private FakeCommandRepository commands; + private FakeExecutionRepository executions; + private FakeEventRepository events; + private FakeReservationRepository reservations; + private FakeRecoveryCaseOpener recovery; + private FeedbackReconciler reconciler; + + @BeforeEach + void setUp() { + commands = new FakeCommandRepository(); + executions = new FakeExecutionRepository(); + events = new FakeEventRepository(); + reservations = new FakeReservationRepository(); + recovery = new FakeRecoveryCaseOpener(); + reconciler = new FeedbackReconciler(commands, executions, events, + reservations, recovery); + + DeviceCommand command = DeviceCommand.pending( + "CMD-1", "CMD-KEY-1", "VEH-1", DeviceCommandType.MOVE_ROUTE, + "{}", 1, at(0)); + command.markSent("SEND-1", at(1)); + commands.put(command); + executions.put(execution()); + commands.resetUpdateCount(); + } + + @Test + void duplicateEventDoesNotAdvanceTheCommandTwice() { + DeviceEvent completed = new DeviceEvent( + "E-1", "VEH-1", "CMD-KEY-1", 10L, at(10), at(11), + DeviceEvent.Type.COMPLETED, "TASK-1", "C", + Collections.singletonList("C")); + + EventAcceptanceResult first = reconciler.accept(completed); + EventAcceptanceResult duplicate = reconciler.accept(completed); + + assertEquals(EventAcceptanceResult.APPLIED, first); + assertEquals(EventAcceptanceResult.DUPLICATE, duplicate); + assertEquals(1, commands.updateCount("CMD-1")); + assertEquals(1, reservations.releaseCallCount); + } + + @Test + void taskNumberMismatchCreatesRecoveryInsteadOfReleasing() { + DeviceSnapshot snapshot = snapshot("WRONG-TASK", "B", + Arrays.asList("B", "C"), 5L, false); + + ReconcileResult result = reconciler.reconcile(snapshot); + + assertEquals(ReconcileStatus.RECOVERY_REQUIRED, result.getStatus()); + assertEquals(0, reservations.releaseCallCount); + assertEquals(Collections.singletonList("TASK_NUMBER_MISMATCH"), + recovery.reasons); + } + + @Test + void releasesOnlyResourcesStrictlyPassedByMatchingProgress() { + DeviceSnapshot snapshot = snapshot("TASK-1", "B", + Arrays.asList("B", "C"), 5L, false); + + ReconcileResult result = reconciler.reconcile(snapshot); + + assertEquals(ReconcileStatus.ADVANCED, result.getStatus()); + assertEquals(Arrays.asList("NODE:A", "EDGE:A:B"), + result.getReleasedResourceKeys()); + assertEquals(Arrays.asList("NODE:A", "EDGE:A:B"), + reservations.releasedResources); + } + + @Test + void staleSnapshotIsIgnoredWithoutReleasingAgain() { + reconciler.reconcile(snapshot("TASK-1", "B", + Arrays.asList("B", "C"), 5L, false)); + + ReconcileResult stale = reconciler.reconcile(snapshot("TASK-1", "C", + Collections.singletonList("C"), 4L, false)); + + assertEquals(ReconcileStatus.IGNORED_STALE, stale.getStatus()); + assertEquals(1, reservations.releaseCallCount); + } + + @Test + void unknownCommandUsesNonTerminalSnapshotForDiagnosticsOnly() { + DeviceCommand command = commands.findByCommandKey("CMD-KEY-1"); + command.markUnknown("WRITE_RESULT_AMBIGUOUS", at(2)); + commands.save(command); + commands.resetUpdateCount(); + + ReconcileResult result = reconciler.reconcile(snapshot("TASK-1", "B", + Arrays.asList("B", "C"), 5L, false)); + + assertEquals(ReconcileStatus.CONSISTENT, result.getStatus()); + assertTrue(result.getReleasedResourceKeys().isEmpty()); + assertEquals(0, reservations.releaseCallCount); + assertEquals(-1L, executions.findByCommandKey("CMD-KEY-1") + .getLastDeviceSequence()); + assertEquals(DeviceCommandState.UNKNOWN, command.getState()); + } + + @Test + void terminalCommandCannotReleaseResourcesFromLateCompletion() { + DeviceCommand command = commands.findByCommandKey("CMD-KEY-1"); + command.reject("PLC_REJECTED", at(2)); + commands.save(command); + commands.resetUpdateCount(); + + ReconcileResult result = reconciler.reconcile(snapshot("TASK-1", "C", + Collections.singletonList("C"), 5L, true)); + + assertEquals(ReconcileStatus.RECOVERY_REQUIRED, result.getStatus()); + assertEquals(0, reservations.releaseCallCount); + assertEquals(-1L, executions.findByCommandKey("CMD-KEY-1") + .getLastDeviceSequence()); + } + + @Test + void acknowledgementDoesNotRequireRouteFacts() { + DeviceEvent acknowledgement = new DeviceEvent( + "ACK-1", "VEH-1", "CMD-KEY-1", 2L, at(2), at(2), + DeviceEvent.Type.ACKNOWLEDGED, "TASK-1", null, null); + + assertEquals(EventAcceptanceResult.APPLIED, + reconciler.accept(acknowledgement)); + assertEquals(DeviceCommandState.ACKNOWLEDGED, + commands.findByCommandKey("CMD-KEY-1").getState()); + assertEquals(0, reservations.releaseCallCount); + } + + private static CommittedDeviceExecution execution() { + return new CommittedDeviceExecution( + "CMD-1", "CMD-KEY-1", "VEH-1", "TASK-1", + Arrays.asList("A", "B", "C"), + Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B", + "EDGE:B:C", "NODE:C"), + 7L, 11L, "BOOT-1"); + } + + private static DeviceSnapshot snapshot(String taskNumber, String currentNode, + List remaining, long sequence, + boolean completed) { + return new DeviceSnapshot("SNAP-" + sequence, "VEH-1", + "CMD-KEY-1", taskNumber, sequence, at(sequence), true, + "BOOT-1", 7L, 11L, currentNode, remaining, completed); + } + + private static Instant at(long seconds) { + return Instant.EPOCH.plusSeconds(seconds); + } + + private static final class FakeCommandRepository + implements DeviceCommandRepository { + private final Map byId = + new HashMap(); + private final Map updates = + new HashMap(); + + void put(DeviceCommand command) { + byId.put(command.getCommandId(), command); + } + + void resetUpdateCount() { + updates.clear(); + } + + int updateCount(String commandId) { + Integer count = updates.get(commandId); + return count == null ? 0 : count; + } + + @Override + public DeviceCommand findByCommandKey(String commandKey) { + for (DeviceCommand command : byId.values()) { + if (command.getCommandKey().equals(commandKey)) { + return command; + } + } + return null; + } + + @Override + public DeviceCommand findByCommandId(String commandId) { + return byId.get(commandId); + } + + @Override + public void save(DeviceCommand command) { + byId.put(command.getCommandId(), command); + Integer count = updates.get(command.getCommandId()); + updates.put(command.getCommandId(), count == null ? 1 : count + 1); + } + } + + private static final class FakeExecutionRepository + implements DeviceExecutionRepository { + private final Map byCommand = + new HashMap(); + + void put(CommittedDeviceExecution execution) { + byCommand.put(execution.getCommandKey(), execution); + } + + @Override + public CommittedDeviceExecution findByCommandKey(String commandKey) { + return byCommand.get(commandKey); + } + + @Override + public CommittedDeviceExecution findActiveByDeviceId(String deviceId) { + for (CommittedDeviceExecution execution : byCommand.values()) { + if (execution.getDeviceId().equals(deviceId)) { + return execution; + } + } + return null; + } + + @Override + public void save(CommittedDeviceExecution execution) { + put(execution); + } + } + + private static final class FakeEventRepository + implements DeviceEventRepository { + private final Set ids = new HashSet(); + + @Override + public boolean claim(DeviceEvent event) { + return ids.add(event.getEventId()); + } + } + + private static final class FakeReservationRepository + implements ResourceReservationRepository { + private int releaseCallCount; + private List releasedResources = Collections.emptyList(); + + @Override + public void releasePassedResources(String commandKey, + List resourceKeys, + String releaseEvidenceId, + long fencingToken) { + releaseCallCount++; + releasedResources = new ArrayList(resourceKeys); + } + + @Override + public void protectForUnknown(String commandKey, String evidenceId) { + } + } + + private static final class FakeRecoveryCaseOpener + implements RecoveryCaseOpener { + private final List reasons = new ArrayList(); + + @Override + public void open(String deviceId, String commandKey, String reason, + List evidenceIds) { + reasons.add(reason); + assertTrue(!evidenceIds.isEmpty()); + } + } +}