From 3d3507a770cc067209cfb779840a1a01ae5b87af Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Tue, 13 Oct 2026 23:40:44 +0800 Subject: [PATCH] feat(selfop): add bounded independent quantity diagnostic capture --- .../LingxingQuantityCaptureProperties.java | 19 ++ .../lingxing/LingxingQuantityCapture.java | 171 ++++++++++++++++++ .../lingxing/LingxingQuantityEvidence.java | 34 ++++ .../lingxing/LingxingQuantityHttpSink.java | 57 ++++++ .../provider/lingxing/LingxingWmsClient.java | 15 +- .../lingxing/LingxingQuantityCaptureTest.java | 146 +++++++++++++++ .../LingxingQuantityHttpSinkTest.java | 65 +++++++ .../lingxing/LingxingWmsClientTest.java | 18 ++ .../QuantityDiagnosticConfiguration.java | 8 + .../QuantityDiagnosticContract.java | 22 +++ .../QuantityDiagnosticController.java | 52 ++++++ .../diagnostic/QuantityDiagnosticPolicy.java | 89 +++++++++ .../QuantityDiagnosticProperties.java | 11 ++ .../diagnostic/QuantityDiagnosticService.java | 125 +++++++++++++ .../db/selfop-quantity-diagnostics.sql | 38 ++++ ...lfop-quantity-diagnostics-internal-v1.yaml | 105 +++++++++++ .../QuantityDiagnosticControllerTest.java | 47 +++++ .../QuantityDiagnosticServiceTest.java | 169 +++++++++++++++++ 18 files changed, 1190 insertions(+), 1 deletion(-) create mode 100644 integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java create mode 100644 integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java create mode 100644 integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java create mode 100644 integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java create mode 100644 integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java create mode 100644 integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticConfiguration.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java create mode 100644 lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql create mode 100644 lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java diff --git a/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java b/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java new file mode 100644 index 000000000..1e94414c2 --- /dev/null +++ b/integration-service/src/main/java/com/baiyue/integration/config/LingxingQuantityCaptureProperties.java @@ -0,0 +1,19 @@ +package com.baiyue.integration.config; + +import java.time.Instant; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; + +/** Fixed one-shot scope. No refresh/restart may extend the configured absolute window. */ +@ConfigurationProperties("baiyue.integration.lingxing-wms.quantity-capture") +public record LingxingQuantityCaptureProperties(@DefaultValue("false") boolean enabled, + String captureId, String sourceAccount, String rootTaskId, String sessionId, + String warehouseCode, String customerCode, Instant startsAt, Instant expiresAt) { + public boolean valid() { + return text(captureId) && text(sourceAccount) && text(rootTaskId) && text(sessionId) + && text(warehouseCode) && text(customerCode) && startsAt != null && expiresAt != null + && expiresAt.isAfter(startsAt) && !expiresAt.isAfter(startsAt.plusSeconds(1800)); + } + private static boolean text(String value) { return value != null && !value.isBlank() && value.length() <= 128; } + @Override public String toString() { return "LingxingQuantityCaptureProperties[enabled=" + enabled + "]"; } +} diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java new file mode 100644 index 000000000..8500b810c --- /dev/null +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCapture.java @@ -0,0 +1,171 @@ +package com.baiyue.integration.provider.lingxing; + +import com.baiyue.integration.config.LingxingQuantityCaptureProperties; +import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.EgressContext; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.annotation.PreDestroy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import java.math.BigDecimal; +import java.time.Clock; +import java.time.Instant; +import java.util.*; +import java.util.concurrent.*; +import static com.baiyue.integration.provider.lingxing.LingxingQuantityEvidence.*; + +/** Passive bounded observer. It has no supplier client and never changes collection decisions. */ +@Component +public final class LingxingQuantityCapture { + private static final Logger LOG = LoggerFactory.getLogger(LingxingQuantityCapture.class); + private static final int MAX_ORDERS = 100, MAX_ROWS = 2000, MAX_BYTES = 2 * 1024 * 1024; + private final LingxingQuantityCaptureProperties config; + private final ObjectMapper mapper; + private final Sink sink; + private final Clock clock; + private final ThreadPoolExecutor worker; + private final Map pending = new HashMap<>(); + private final Set completed = new HashSet<>(); + private volatile String stopReason; + private int records, rows, bytes; + + @Autowired + public LingxingQuantityCapture(LingxingQuantityCaptureProperties config, ObjectMapper mapper, + ObjectProvider sink) { + this(config, mapper, sink.getIfAvailable(), Clock.systemUTC()); + } + LingxingQuantityCapture(LingxingQuantityCaptureProperties config, ObjectMapper mapper, Sink sink, Clock clock) { + this.config = config; this.mapper = mapper; this.sink = sink; this.clock = clock; + worker = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(MAX_ORDERS), task -> { + Thread thread = new Thread(task, "lingxing-quantity-capture"); thread.setDaemon(true); return thread; + }, new ThreadPoolExecutor.AbortPolicy()); + if (!config.enabled()) stopReason = "DISABLED"; + else if (!config.valid() || sink == null) stopReason = "CONFIGURATION_UNAVAILABLE"; + } + + /** Called only after the normal projection has successfully validated a supplier response. */ + public synchronized void observe(JsonNode data, boolean detail, EgressContext context, Instant observedAt) { + try { + if (!active() || context == null || !context.wellFormed() + || !Objects.equals(config.sourceAccount(), context.sourceAccount()) + || !Objects.equals(config.rootTaskId(), context.rootTaskId()) + || !Objects.equals(config.sessionId(), context.sessionId())) return; + if (detail) enqueue(data, true, context, observedAt); + else for (JsonNode row : data.path("records")) { + if (!active()) break; + enqueue(row, false, context, observedAt); + } + } catch (RuntimeException error) { stop("OBSERVATION_FAILED"); } + } + private void enqueue(JsonNode data, boolean detail, EgressContext context, Instant time) { + String warehouse = text(data, "whCode", 64), customer = text(data, "customerCode", 64); + if (!("*".equals(config.warehouseCode()) || config.warehouseCode().equals(warehouse)) + || !("*".equals(config.customerCode()) || config.customerCode().equals(customer))) return; + Key key = new Key(warehouse, customer, text(data, "deliveryNo", 128)); + if (completed.contains(key)) return; + // Freeze the first 100 list candidates. Extra page rows and unselected details cost no retained memory. + if (detail && !pending.containsKey(key)) return; + if (!detail && (pending.containsKey(key) || pending.size() + completed.size() >= MAX_ORDERS)) return; + JsonNode items = data.path(detail ? "skuVOList" : "productList"); + if (!items.isArray() || items.isEmpty() || items.size() > 200) { stop("SKU_RECORD_LIMIT_OR_MISSING"); return; } + Map quantities = new TreeMap<>(); + for (JsonNode item : items) { + String sku = text(item, "productSku", 128); + Quantities next = new Quantities(detail ? null : quantity(item.get("qty")), + detail ? quantity(item.get("deliveryQty")) : null, + detail ? quantity(item.get("actualDeliveryQty")) : null); + quantities.merge(sku, next, Quantities::add); + } + Observation observation = new Observation(data.path("status").isInt() ? data.path("status").asText() : null, + optionalText(data, "outboundTime", 64), time, context.attemptId(), true); + Sample sample = new Sample(key, detail, observation, quantities, items.size()); + process(sample); + } + private void process(Sample sample) { + try { + if (!active()) return; + if (++records > 2000 || (rows += sample.rowCount()) > MAX_ROWS + || (bytes += mapper.writeValueAsBytes(sample).length) > MAX_BYTES) { stop("CAPTURE_LIMIT"); return; } + if (completed.contains(sample.key())) return; + // Directory and detail stages can be far apart: candidates live until the fixed session expiry. + Pending pair = pending.computeIfAbsent(sample.key(), ignored -> new Pending()); + // Keep the first observation on each side; retries cannot silently replace its generation. + if (sample.detail()) { if (pair.detail == null) pair.detail = sample; } + else if (pair.list == null) pair.list = sample; + if (pair.list == null || pair.detail == null) return; + Set skus = new TreeSet<>(pair.list.quantities().keySet()); skus.addAll(pair.detail.quantities().keySet()); + if (skus.size() > 200) { stop("ORDER_SKU_LIMIT"); return; } + List lines = new ArrayList<>(); boolean difference = false; + for (String sku : skus) { + Quantities list = pair.list.quantities().get(sku), detail = pair.detail.quantities().get(sku); + BigDecimal forecast = detail == null ? null : detail.delivery, actual = detail == null ? null : detail.actual; + difference |= forecast != null && actual != null && forecast.compareTo(actual) != 0; + lines.add(new Line(sku, list != null, detail != null, list == null ? null : list.list, forecast, actual)); + } + Key key = sample.key(); + LingxingQuantityEvidence evidence = new LingxingQuantityEvidence(config.captureId(), config.sourceAccount(), + config.rootTaskId(), config.sessionId(), key.warehouse(), key.customer(), key.order(), + pair.list.observation(), pair.detail.observation(), List.copyOf(lines)); + worker.execute(() -> { + try { + if (!clock.instant().isBefore(config.expiresAt())) { abortTransmission("EXPIRED"); return; } + sink.accept(evidence); + } catch (ReceiverStopped normalStop) { abortTransmission(normalStop.reason()); } + catch (Exception failure) { abortTransmission("DIAGNOSTIC_FAILED"); } + }); + completed.add(key); pending.remove(key); + if (difference) stop("FORECAST_ACTUAL_DIFFERENCE"); + else if (completed.size() >= MAX_ORDERS) stop("ORDER_LIMIT"); + } catch (Exception error) { stop("DIAGNOSTIC_FAILED"); } + } + private boolean active() { + if (stopReason != null) return false; + if (!clock.instant().isBefore(config.expiresAt())) { stop("EXPIRED"); return false; } + return !clock.instant().isBefore(config.startsAt()); + } + private synchronized void stop(String reason) { + if (stopReason != null) return; + stopReason = reason; + if (!"ORDER_LIMIT".equals(reason) && !"FORECAST_ACTUAL_DIFFERENCE".equals(reason)) worker.getQueue().clear(); + pending.clear(); + LOG.info("Lingxing quantity capture stopped: {}", reason); + } + private synchronized void abortTransmission(String reason) { + worker.getQueue().clear(); pending.clear(); + if (!reason.equals(stopReason)) { + stopReason = reason; LOG.info("Lingxing quantity capture stopped: {}", reason); + } + } + public String stopReason() { return stopReason; } + @PreDestroy public void close() { stop("SHUTDOWN"); worker.shutdownNow(); } + private static String text(JsonNode data, String field, int max) { + String value = optionalText(data, field, max); + if (value == null || value.isBlank()) throw new IllegalArgumentException("Missing diagnostic identity"); + return value; + } + private static String optionalText(JsonNode data, String field, int max) { + JsonNode value = data.get(field); + if (value == null || value.isNull()) return null; + if (!value.isTextual() || value.textValue().length() > max) throw new IllegalArgumentException("Invalid diagnostic text"); + return value.textValue(); + } + private static BigDecimal quantity(JsonNode value) { + if (value == null || value.isNull() || value.isTextual() && value.textValue().isBlank()) return null; + if (!value.isNumber()) throw new IllegalArgumentException("Invalid diagnostic quantity"); + BigDecimal result = value.decimalValue(); + if (result.precision() > 30 || Math.abs(result.scale()) > 10) throw new IllegalArgumentException("Quantity limit"); + return result; + } + private record Key(String warehouse, String customer, String order) { } + private record Quantities(BigDecimal list, BigDecimal delivery, BigDecimal actual) { + Quantities add(Quantities other) { return new Quantities(sum(list, other.list), sum(delivery, other.delivery), sum(actual, other.actual)); } + private static BigDecimal sum(BigDecimal a, BigDecimal b) { return a == null || b == null ? null : a.add(b); } + } + private record Sample(Key key, boolean detail, Observation observation, Map quantities, int rowCount) { } + private static final class Pending { + private Sample list, detail; + } +} diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java new file mode 100644 index 000000000..0e871d723 --- /dev/null +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityEvidence.java @@ -0,0 +1,34 @@ +package com.baiyue.integration.provider.lingxing; + +import java.math.BigDecimal; +import java.time.Instant; +import java.util.List; + +/** Whitelist only; never includes source payload, addresses, labels or credentials. */ +public record LingxingQuantityEvidence(String diagnosticSessionId, String sourceAccount, String rootTaskId, + String collectionSessionId, String warehouseCode, String ownerCode, String orderNo, + Observation listObservation, Observation detailObservation, List skus) { + public record Observation(String status, String outboundTime, Instant observedAt, String attemptId, + boolean skuSetComplete) { } + public record Line(String sku, boolean listPresent, boolean detailPresent, BigDecimal listQty, + BigDecimal forecastQty, BigDecimal actualQty) { } + public boolean skuSetEqual() { return skus.stream().allMatch(line -> line.listPresent() && line.detailPresent()); } + public boolean forecastActualDifference() { return skus.stream().anyMatch(line -> line.forecastQty() != null + && line.actualQty() != null && line.forecastQty().compareTo(line.actualQty()) != 0); } + /** Receiver must durably enforce window, deduplication and cumulative bounds across restarts. */ + public interface Sink { void accept(LingxingQuantityEvidence evidence); } + /** Normal durable receiver decision, including an accepted final order; not a transport failure. */ + public static final class ReceiverStopped extends RuntimeException { + private final String reason; + private final boolean orderStored; + public ReceiverStopped(String reason, boolean orderStored) { + super("Diagnostic receiver stopped"); + if (reason == null || !reason.matches("[A-Z][A-Z0-9_]{0,63}")) { + throw new IllegalArgumentException("Invalid diagnostic stop reason"); + } + this.reason = reason; this.orderStored = orderStored; + } + public String reason() { return reason; } + public boolean orderStored() { return orderStored; } + } +} diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java new file mode 100644 index 000000000..0a5571b22 --- /dev/null +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSink.java @@ -0,0 +1,57 @@ +package com.baiyue.integration.provider.lingxing; + +import com.baiyue.integration.config.LingxingEgressProperties; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import java.net.URI; +import java.net.http.*; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Set; + +/** Async worker calls this once per paired order. No retry and no supplier requests. */ +@Component +public final class LingxingQuantityHttpSink implements LingxingQuantityEvidence.Sink { + private static final String PATH = "/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders"; + private final URI baseUri; + private final ObjectMapper mapper; + private final String token; + private final HttpClient client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)) + .followRedirects(HttpClient.Redirect.NEVER).build(); + public LingxingQuantityHttpSink(LingxingEgressProperties config, ObjectMapper mapper, + @Value("${baiyue.security.internal-service-token:}") String token) { + this.baseUri = config.budgetAuthorityBaseUri(); + this.mapper = mapper.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + this.token = token; + } + @Override public void accept(LingxingQuantityEvidence evidence) { + try { + if (baseUri == null || token == null || token.getBytes(StandardCharsets.UTF_8).length < 32) throw unavailable(); + byte[] body = mapper.writeValueAsBytes(evidence); + if (body.length > 256 * 1024 || evidence.skus().size() > 200) throw unavailable(); + HttpRequest request = HttpRequest.newBuilder(baseUri.resolve(PATH)).timeout(Duration.ofSeconds(3)) + .header("X-Internal-Service-Token", token).header("Content-Type", "application/json; charset=UTF-8") + .POST(HttpRequest.BodyPublishers.ofByteArray(body)).build(); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); + if (response.statusCode() != 200 || response.body().length() > 16384) throw unavailable(); + JsonNode root = mapper.readTree(response.body()); + if (root == null || !"200".equals(root.path("code").textValue())) throw unavailable(); + String outcome = root.path("data").path("outcome").asText(); + if (!Set.of("ACCEPTED", "DUPLICATE", "STOPPED").contains(outcome)) throw unavailable(); + // Receiver may accept the final order and atomically close its durable session. + JsonNode reason = root.path("data").path("stopReason"); + if (!reason.isMissingNode() && !reason.isNull()) { + if (!reason.isTextual() || !reason.textValue().matches("[A-Z][A-Z0-9_]{0,63}")) throw unavailable(); + throw new LingxingQuantityEvidence.ReceiverStopped(reason.textValue(), !"STOPPED".equals(outcome)); + } + if ("STOPPED".equals(outcome)) throw unavailable(); + } catch (LingxingQuantityEvidence.ReceiverStopped normalStop) { + throw normalStop; + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); throw unavailable(); + } catch (Exception failed) { throw unavailable(); } + } + private static IllegalStateException unavailable() { return new IllegalStateException("DIAGNOSTIC_RECEIVER_UNAVAILABLE"); } +} diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java index fda1731da..33f3d4373 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingWmsClient.java @@ -48,6 +48,12 @@ public final class LingxingWmsClient { private final HttpClient httpClient; private final Clock clock; private final LingxingEgressGate egressGate; + private LingxingQuantityCapture quantityCapture; + + @Autowired + void setQuantityCapture(LingxingQuantityCapture quantityCapture) { + this.quantityCapture = quantityCapture; + } @Autowired public LingxingWmsClient(LingxingWmsProperties properties, ObjectMapper mapper, LingxingEgressGate egressGate) { @@ -184,7 +190,14 @@ public final class LingxingWmsClient { } throw error; } - return projection.apply(json, handle.metadata()); + T result = projection.apply(json, handle.metadata()); + if (quantityCapture != null && (operation == Operation.OUTBOUND_QUERY + || operation == Operation.OUTBOUND_LOCATE || operation == Operation.OUTBOUND_DETAIL)) { + // Diagnostics must never alter the normal projection or the egress completion result. + try { quantityCapture.observe(json, operation == Operation.OUTBOUND_DETAIL, context, clock.instant()); } + catch (RuntimeException ignored) { /* Passive diagnostic failure is isolated. */ } + } + return result; }); } diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java new file mode 100644 index 000000000..7624cd4d5 --- /dev/null +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityCaptureTest.java @@ -0,0 +1,146 @@ +package com.baiyue.integration.provider.lingxing; + +import com.baiyue.integration.config.LingxingQuantityCaptureProperties; +import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.*; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.time.*; +import java.util.concurrent.*; +import static org.assertj.core.api.Assertions.*; + +class LingxingQuantityCaptureTest { + private static final Instant NOW = Instant.parse("2026-10-13T01:00:00Z"); + private final ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule()); + private final BlockingQueue saved = new LinkedBlockingQueue<>(); + private EgressContext context(String session) { return new EgressContext("account", CollectionMode.BUSINESS, + "root", session, "attempt", "grant"); } + private LingxingQuantityCapture capture(Instant expiry, LingxingQuantityEvidence.Sink sink) { + return new LingxingQuantityCapture(new LingxingQuantityCaptureProperties(true, "capture", "account", + "root", "session", "*", "*", NOW.minusSeconds(1), expiry), mapper, sink, Clock.fixed(NOW, ZoneOffset.UTC)); + } + private void observe(LingxingQuantityCapture c, boolean detail, String items, String session, int status) throws Exception { + String order = "{\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"D\",\"status\":" + status + + ",\"" + (detail ? "skuVOList" : "productList") + "\":[" + items + "]}"; + c.observe(mapper.readTree(detail ? order : "{\"records\":[" + order + "]}"), detail, context(session), NOW); + } + @ParameterizedTest @ValueSource(ints = {10, 8}) + void preservesBothSemanticOutcomesAndStopsAtDifference(int listQty) throws Exception { + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { + observe(c.value, false, "{\"productSku\":\"S\",\"qty\":" + listQty + "}", "session", 100); + observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":10,\"actualDeliveryQty\":8}", "session", 100); + var e = saved.poll(3, TimeUnit.SECONDS); + assertThat(e).isNotNull(); assertThat(e.forecastActualDifference()).isTrue(); + assertThat(e.skus().get(0).listQty()).isEqualByComparingTo(Integer.toString(listQty)); + assertThat(e.skuSetEqual()).isTrue(); + awaitStop(c.value, "FORECAST_ACTUAL_DIFFERENCE"); + } + } + @Test void duplicateDecimalsUnknownsAndSkuMismatchRemainVisibleForCancelledOrder() throws Exception { + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { + observe(c.value, false, "{\"productSku\":\"S\",\"qty\":0.1},{\"productSku\":\"S\",\"qty\":0.2}," + + "{\"productSku\":\"LIST_ONLY\",\"qty\":1}", "session", 99); + observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":null}," + + "{\"productSku\":\"S\",\"deliveryQty\":2,\"actualDeliveryQty\":1}", "session", 99); + var e = saved.poll(3, TimeUnit.SECONDS); + assertThat(e).isNotNull(); assertThat(e.skuSetEqual()).isFalse(); + var line = e.skus().stream().filter(v -> v.sku().equals("S")).findFirst().orElseThrow(); + assertThat(line.listQty()).isEqualByComparingTo("0.3"); + assertThat(line.forecastQty()).isEqualByComparingTo("3"); assertThat(line.actualQty()).isNull(); + assertThat(e.detailObservation().status()).isEqualTo("99"); assertThat(e.detailObservation().outboundTime()).isNull(); + assertThat(e.forecastActualDifference()).isFalse(); + } + } + @Test void crossSessionCannotPairAndExpiredCaptureDoesNotWrite() throws Exception { + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { + observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "other", 100); + observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}", "session", 100); + assertThat(saved.poll(100, TimeUnit.MILLISECONDS)).isNull(); + } + try (var c = new CloseableCapture(capture(NOW, saved::add))) { + observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100); + assertThat(c.value.stopReason()).isEqualTo("EXPIRED"); assertThat(saved).isEmpty(); + } + } + @Test void sinkFailureStopsOnlyDiagnosticAndNeverEscapesCaller() throws Exception { + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { throw new IllegalStateException("private"); }))) { + observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100); + observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}", "session", 100); + awaitStop(c.value, "DIAGNOSTIC_FAILED"); + } + } + @Test void hundredCompleteOrdersStopsFurtherCapture() throws Exception { + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) { + for (int i = 0; i < 100; i++) { + String head = "\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"D" + i + "\",\"status\":100,"; + c.value.observe(mapper.readTree("{\"records\":[{" + head + "\"productList\":[{\"productSku\":\"S\",\"qty\":1}]}]}"), false, context("session"), NOW); + c.value.observe(mapper.readTree("{" + head + "\"skuVOList\":[{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}]}"), true, context("session"), NOW); + assertThat(saved.poll(3, TimeUnit.SECONDS)).isNotNull(); + } + awaitStop(c.value, "ORDER_LIMIT"); + observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100); + assertThat(saved).isEmpty(); + } + } + @Test void thousandRowPageSelectsHundredAndSlowSinkDoesNotBlockDetails() throws Exception { + CountDownLatch entered = new CountDownLatch(1), release = new CountDownLatch(1); + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { + entered.countDown(); + try { release.await(3, TimeUnit.SECONDS); } catch (InterruptedException error) { Thread.currentThread().interrupt(); } + saved.add(evidence); + }))) { + var page = mapper.createObjectNode(); var records = page.putArray("records"); + for (int i = 0; i < 1000; i++) { + var order = records.addObject().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100); + order.putArray("productList").addObject().put("productSku", "S").put("qty", 1); + } + c.value.observe(page, false, context("session"), NOW); + assertThat(c.value.stopReason()).isNull(); + // Unselected details arrive first: they must not consume queue or evidence capacity. + for (int i = 999; i >= 0; i--) { + var detail = mapper.createObjectNode().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100); + detail.putArray("skuVOList").addObject().put("productSku", "S").put("deliveryQty", 1).put("actualDeliveryQty", 1); + c.value.observe(detail, true, context("session"), NOW); + } + assertThat(entered.await(3, TimeUnit.SECONDS)).isTrue(); + assertThat(c.value.stopReason()).isEqualTo("ORDER_LIMIT"); + release.countDown(); + for (int i = 0; i < 100; i++) assertThat(saved.poll(3, TimeUnit.SECONDS)).isNotNull(); + assertThat(saved).isEmpty(); + } finally { release.countDown(); } + } + private static void awaitStop(LingxingQuantityCapture capture, String reason) throws Exception { + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); + while (capture.stopReason() == null && System.nanoTime() < end) Thread.sleep(5); + assertThat(capture.stopReason()).isEqualTo(reason); + } + @ParameterizedTest @ValueSource(strings = {"FORECAST_ACTUAL_DIFFERENCE", "LIMIT_REACHED"}) + void savedFinalOrderRetainsNormalReceiverReasonAndDiscardsQueuedOrders(String reason) throws Exception { + CountDownLatch entered = new CountDownLatch(1), release = new CountDownLatch(1); + try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { + entered.countDown(); + try { release.await(3, TimeUnit.SECONDS); } catch (InterruptedException error) { Thread.currentThread().interrupt(); } + saved.add(evidence); // Receiver persisted the last accepted order before returning its stop decision. + throw new LingxingQuantityEvidence.ReceiverStopped(reason, true); + }))) { + for (int i = 0; i < 3; i++) { + var order = mapper.createObjectNode().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100); + order.putArray("productList").addObject().put("productSku", "S").put("qty", 1); + var page = mapper.createObjectNode(); page.putArray("records").add(order); + c.value.observe(page, false, context("session"), NOW); + order.putArray("skuVOList").addObject().put("productSku", "S").put("deliveryQty", 1).put("actualDeliveryQty", 1); + c.value.observe(order, true, context("session"), NOW); + } + assertThat(entered.await(3, TimeUnit.SECONDS)).isTrue(); + release.countDown(); awaitStop(c.value, reason); + assertThat(saved).hasSize(1); + assertThat(saved.remove().orderNo()).isEqualTo("D0"); + assertThat(saved.poll(100, TimeUnit.MILLISECONDS)).isNull(); + } finally { release.countDown(); } + } + private record CloseableCapture(LingxingQuantityCapture value) implements AutoCloseable { + @Override public void close() { value.close(); } + } +} diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java new file mode 100644 index 000000000..6fb44e246 --- /dev/null +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingQuantityHttpSinkTest.java @@ -0,0 +1,65 @@ +package com.baiyue.integration.provider.lingxing; + +import com.baiyue.integration.config.LingxingEgressProperties; +import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.CollectionMode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.Test; +import java.net.*; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import static org.assertj.core.api.Assertions.*; + +class LingxingQuantityHttpSinkTest { + @Test void exactContractAuthenticationAndNoRetryWhenReceiverStops() throws Exception { + var calls = new AtomicInteger(); var body = new AtomicReference(); var header = new AtomicReference(); + var outcome = new AtomicReference<>("ACCEPTED"); + var reason = new AtomicReference<>("null"); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders", exchange -> { + calls.incrementAndGet(); body.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + header.set(exchange.getRequestHeaders().getFirst("X-Internal-Service-Token")); + byte[] response = ("{\"code\":\"200\",\"data\":{\"outcome\":\"" + outcome.get() + "\",\"stopReason\":" + reason.get() + "}}") + .getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, response.length); exchange.getResponseBody().write(response); exchange.close(); + }); server.start(); + try { + var props = new LingxingEgressProperties(CollectionMode.OFF, "test", null, null, null, + URI.create("http://127.0.0.1:" + server.getAddress().getPort()), null, null, false); + var mapper = new ObjectMapper().registerModule(new JavaTimeModule()) + .disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + String token = "synthetic-internal-token-000000000000"; + var sink = new LingxingQuantityHttpSink(props, mapper, token); + var obs = new LingxingQuantityEvidence.Observation("99", null, Instant.parse("2026-10-13T00:00:00Z"), "attempt", true); + var evidence = new LingxingQuantityEvidence("diagnostic", "account", "root", "collection", "W", "C", "D", obs, obs, + List.of(new LingxingQuantityEvidence.Line("S", true, true, null, null, null))); + sink.accept(evidence); + var json = mapper.readTree(body.get()); + assertThat(header.get()).isEqualTo(token); assertThat(json.path("diagnosticSessionId").asText()).isEqualTo("diagnostic"); + assertThat(json.path("listObservation").path("observedAt").asText()).isEqualTo("2026-10-13T00:00:00Z"); + assertThat(json.path("skus").get(0).path("forecastQty").isNull()).isTrue(); + assertThat(json.has("skuSetEqual")).isFalse(); assertThat(json.has("forecastActualDifference")).isFalse(); + outcome.set("STOPPED"); assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOf(IllegalStateException.class); + assertThat(calls.get()).isEqualTo(2); + for (String stopReason : List.of("FORECAST_ACTUAL_DIFFERENCE", "LIMIT_REACHED")) { + outcome.set("ACCEPTED"); reason.set("\"" + stopReason + "\""); + assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOfSatisfying( + LingxingQuantityEvidence.ReceiverStopped.class, stopped -> { + assertThat(stopped.reason()).isEqualTo(stopReason); + assertThat(stopped.orderStored()).isTrue(); + }); + } + outcome.set("STOPPED"); reason.set("\"DISABLED\""); + assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOfSatisfying( + LingxingQuantityEvidence.ReceiverStopped.class, stopped -> { + assertThat(stopped.reason()).isEqualTo("DISABLED"); + assertThat(stopped.orderStored()).isFalse(); + }); + assertThat(calls.get()).isEqualTo(5); + } finally { server.stop(0); } + } +} diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java index e2d42ba2c..29a4dd64a 100644 --- a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingWmsClientTest.java @@ -52,6 +52,24 @@ class LingxingWmsClientTest { server.start(); } @AfterEach void stop() { server.stop(0); } + @Test void observerFailureCannotChangeBusinessResponseOrIssueExtraRequests() { + var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class); + org.mockito.Mockito.doThrow(new IllegalStateException("diagnostic failure")).when(observer) + .observe(org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.anyBoolean(), + org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.any()); + var c = client(); c.setQuantityCapture(observer); + assertThat(c.queryPage(query(1, 1), context()).records()).hasSize(1); + assertThat(calls.get()).isEqualTo(1); + } + @Test void deniedOrRateLimitedRequestsNeverReachObserver() { + var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class); + var disabled = client(CollectionMode.OFF, true, true); disabled.setQuantityCapture(observer); + assertThrows(LingxingWmsException.class, () -> disabled.queryPage(query(1, 1), context())); + assertThat(calls.get()).isZero(); + var enabled = client(); enabled.setQuantityCapture(observer); status.set(429); + assertThrows(LingxingWmsException.class, () -> enabled.queryPage(query(1, 1), context())); + org.mockito.Mockito.verifyNoInteractions(observer); assertThat(calls.get()).isEqualTo(1); + } @Test void discoveryFiltersOutboundTimeAndUsesWhitelist() throws Exception { var result = client().queryPage(query(1, 1), context()); var data = mapper.readTree(body.get()).path("data"); diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticConfiguration.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticConfiguration.java new file mode 100644 index 000000000..0aef34600 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticConfiguration.java @@ -0,0 +1,8 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(QuantityDiagnosticProperties.class) +public class QuantityDiagnosticConfiguration { } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java new file mode 100644 index 000000000..832548d4e --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java @@ -0,0 +1,22 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import java.math.BigDecimal; +import java.time.Instant; +import java.util.List; + +/** Diagnostic evidence only; never consumed by sales publication. */ +public final class QuantityDiagnosticContract { + private QuantityDiagnosticContract() { } + public record Order(String diagnosticSessionId, String collectionSessionId, String rootTaskId, + String sourceAccount, String warehouseCode, String ownerCode, String orderNo, + Observation listObservation, Observation detailObservation, List skus) { } + public record Observation(String status, String outboundTime, Instant observedAt, String attemptId, + Boolean skuSetComplete) { } + public record Sku(String sku, BigDecimal listQty, BigDecimal forecastQty, BigDecimal actualQty, + Boolean listPresent, Boolean detailPresent) { } + public enum Comparison { + SKU_SET_MISMATCH, QUANTITY_UNKNOWN, FORECAST_ACTUAL_DIFFERENCE, ALL_EQUAL, LIST_QUANTITY_DIFFERENCE + } + public record Receipt(String outcome, Comparison comparison, String stopReason, + int acceptedOrders, int acceptedSkuRows, long acceptedBytes) { } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java new file mode 100644 index 000000000..e189db304 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java @@ -0,0 +1,52 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import com.baiyue.common.core.api.ApiResponse; +import com.baiyue.common.core.api.ResultCode; +import com.baiyue.lastleg.security.InternalServiceTokenGuard; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import org.springframework.dao.DataAccessException; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.transaction.TransactionException; +import org.springframework.web.bind.annotation.*; +import org.springframework.web.server.ResponseStatusException; + +@RestController +@RequestMapping("/internal/lastleg/self-operated-analytics/quantity-diagnostics") +public class QuantityDiagnosticController { + private final InternalServiceTokenGuard guard; + private final QuantityDiagnosticService service; + private final ObjectMapper json; + public QuantityDiagnosticController(InternalServiceTokenGuard guard, QuantityDiagnosticService service, ObjectMapper json) { + this.guard = guard; this.service = service; + this.json = json.copy().enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) + .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + } + @PostMapping(value = "/orders", consumes = "application/json") + public ApiResponse accept( + @RequestHeader(value = "X-Internal-Service-Token", required = false) String token, @RequestBody byte[] body) { + guard.verify(token); + if (body.length > QuantityDiagnosticPolicy.MAX_REQUEST_BYTES) + throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE"); + QuantityDiagnosticContract.Order order; + try { order = json.readValue(body, QuantityDiagnosticContract.Order.class); } + catch (IOException e) { throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_JSON_INVALID"); } + try { return ApiResponse.success(service.accept(order)); } + catch (DataAccessException | TransactionException e) { + // No SQL, request payload or credentials are returned to the caller. + throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "DIAGNOSTIC_STORAGE_UNAVAILABLE"); + } + } + @ExceptionHandler(ResponseStatusException.class) + public ResponseEntity> rejected(ResponseStatusException error) { + ResultCode code = switch (error.getStatusCode().value()) { + case 401 -> ResultCode.UNAUTHORIZED; + case 409 -> ResultCode.CONFLICT; + case 503 -> ResultCode.SERVICE_UNAVAILABLE; + default -> ResultCode.BAD_REQUEST; + }; + return ResponseEntity.status(error.getStatusCode()).body(ApiResponse.failure(code, error.getReason(), null, null)); + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java new file mode 100644 index 000000000..0c1ddd085 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java @@ -0,0 +1,89 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*; +import java.math.BigDecimal; +import java.time.Duration; +import java.time.Instant; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import org.springframework.http.HttpStatus; +import org.springframework.web.server.ResponseStatusException; + +final class QuantityDiagnosticPolicy { + static final int MAX_ORDERS = 100, MAX_ROWS = 2000, MAX_ORDER_ROWS = 200; + static final int MAX_BYTES = 2 * 1024 * 1024, MAX_REQUEST_BYTES = 256 * 1024; + private QuantityDiagnosticPolicy() { } + static ResponseStatusException invalid(String code) { + return new ResponseStatusException(HttpStatus.BAD_REQUEST, code); + } + static void text(String value, int maximum) { + if (value == null || value.isBlank() || value.length() > maximum + || value.codePoints().anyMatch(Character::isISOControl)) throw invalid("DIAGNOSTIC_FIELD_INVALID"); + } + static void settings(QuantityDiagnosticProperties p) { + text(p.diagnosticSessionId(), 128); text(p.collectionSessionId(), 128); + text(p.rootTaskId(), 128); text(p.sourceAccount(), 64); + if (p.startsAt() == null || p.expiresAt() == null || !p.expiresAt().isAfter(p.startsAt()) + || Duration.between(p.startsAt(), p.expiresAt()).compareTo(Duration.ofMinutes(30)) > 0 + || p.warehouseCodes() == null || p.warehouseCodes().isEmpty() || p.warehouseCodes().size() > 20) + throw invalid("DIAGNOSTIC_CONFIGURATION_INVALID"); + p.warehouseCodes().forEach(w -> text(w, 64)); + } + static Order normalize(Order r, QuantityDiagnosticProperties p, Instant now) { + if (r == null) throw invalid("DIAGNOSTIC_ORDER_REQUIRED"); + if (!p.diagnosticSessionId().equals(r.diagnosticSessionId()) + || !p.collectionSessionId().equals(r.collectionSessionId()) || !p.rootTaskId().equals(r.rootTaskId()) + || !p.sourceAccount().equals(r.sourceAccount()) || !p.warehouseCodes().contains(r.warehouseCode())) + throw invalid("DIAGNOSTIC_SCOPE_MISMATCH"); + text(r.ownerCode(), 128); text(r.orderNo(), 128); + observation(r.listObservation(), p, now); observation(r.detailObservation(), p, now); + if (r.skus() == null || r.skus().isEmpty() || r.skus().size() > MAX_ORDER_ROWS) + throw invalid("DIAGNOSTIC_SKU_LIMIT"); + var seen = new HashSet(); + for (Sku s : r.skus()) { + if (s == null) throw invalid("DIAGNOSTIC_SKU_REQUIRED"); + text(s.sku(), 128); + if (!seen.add(s.sku())) throw invalid("DIAGNOSTIC_DUPLICATE_SKU"); + if (s.listPresent() == null || s.detailPresent() == null || (!s.listPresent() && !s.detailPresent())) + throw invalid("DIAGNOSTIC_PRESENCE_INVALID"); + if ((!s.listPresent() && s.listQty() != null) + || (!s.detailPresent() && (s.forecastQty() != null || s.actualQty() != null))) + throw invalid("DIAGNOSTIC_ABSENT_QUANTITY"); + quantity(s.listQty()); quantity(s.forecastQty()); quantity(s.actualQty()); + } + List skus = r.skus().stream().sorted(Comparator.comparing(Sku::sku)) + .map(s -> new Sku(s.sku(), decimal(s.listQty()), decimal(s.forecastQty()), decimal(s.actualQty()), + s.listPresent(), s.detailPresent())).toList(); + return new Order(r.diagnosticSessionId(), r.collectionSessionId(), r.rootTaskId(), r.sourceAccount(), + r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), r.detailObservation(), skus); + } + private static void observation(Observation o, QuantityDiagnosticProperties p, Instant now) { + if (o == null || o.observedAt() == null || o.skuSetComplete() == null) + throw invalid("DIAGNOSTIC_OBSERVATION_REQUIRED"); + text(o.attemptId(), 128); + if (o.status() != null) text(o.status(), 32); + if (o.outboundTime() != null) text(o.outboundTime(), 32); + if (o.observedAt().isBefore(p.startsAt()) || o.observedAt().isAfter(p.expiresAt()) + || o.observedAt().isAfter(now)) throw invalid("DIAGNOSTIC_OBSERVATION_OUTSIDE_SESSION"); + } + private static BigDecimal decimal(BigDecimal v) { return v == null ? null : v.stripTrailingZeros(); } + private static void quantity(BigDecimal v) { + if (v != null && (v.signum() < 0 || v.scale() > 6 || v.precision() - v.scale() > 32)) + throw invalid("DIAGNOSTIC_QUANTITY_INVALID"); + } + static boolean forecastDifference(Order r) { + return r.skus().stream().anyMatch(s -> s.forecastQty() != null && s.actualQty() != null + && s.forecastQty().compareTo(s.actualQty()) != 0); + } + static Comparison compare(Order r) { + if (!r.listObservation().skuSetComplete() || !r.detailObservation().skuSetComplete() + || r.skus().stream().anyMatch(s -> !s.listPresent() || !s.detailPresent())) + return Comparison.SKU_SET_MISMATCH; + if (r.skus().stream().anyMatch(s -> s.listQty() == null || s.forecastQty() == null || s.actualQty() == null)) + return Comparison.QUANTITY_UNKNOWN; + if (forecastDifference(r)) return Comparison.FORECAST_ACTUAL_DIFFERENCE; + return r.skus().stream().allMatch(s -> s.listQty().compareTo(s.actualQty()) == 0) + ? Comparison.ALL_EQUAL : Comparison.LIST_QUANTITY_DIFFERENCE; + } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java new file mode 100644 index 000000000..d65207e9e --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java @@ -0,0 +1,11 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import java.time.Instant; +import java.util.List; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; + +@ConfigurationProperties("baiyue.selfop.quantity-diagnostics") +public record QuantityDiagnosticProperties(@DefaultValue("false") boolean enabled, + String diagnosticSessionId, String collectionSessionId, String rootTaskId, String sourceAccount, + List warehouseCodes, Instant startsAt, Instant expiresAt) { } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java new file mode 100644 index 000000000..c74396862 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java @@ -0,0 +1,125 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*; +import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticPolicy.*; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Clock; +import java.time.Instant; +import java.util.HexFormat; +import java.util.List; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.HttpStatus; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.web.server.ResponseStatusException; + +/** One short transaction per paired order; only the two diagnostic tables are written. */ +@Service +public class QuantityDiagnosticService { + private final JdbcTemplate jdbc; + private final TransactionTemplate tx; + private final ObjectMapper json; + private final QuantityDiagnosticProperties properties; + private final Clock clock; + + @Autowired + public QuantityDiagnosticService(JdbcTemplate jdbc, PlatformTransactionManager manager, + ObjectMapper json, QuantityDiagnosticProperties properties) { + this(jdbc, manager, json, properties, Clock.systemUTC()); + } + QuantityDiagnosticService(JdbcTemplate jdbc, PlatformTransactionManager manager, + ObjectMapper json, QuantityDiagnosticProperties properties, Clock clock) { + this.jdbc = jdbc; this.json = json.copy(); this.properties = properties; this.clock = clock; + this.tx = new TransactionTemplate(manager); this.tx.setTimeout(3); + } + public Receipt accept(Order input) { + if (!properties.enabled()) return new Receipt("STOPPED", null, "DISABLED", 0, 0, 0); + settings(properties); + Order order = normalize(input, properties, clock.instant()); + String payload = encode(order); + int bytes = payload.getBytes(StandardCharsets.UTF_8).length; + if (bytes > MAX_REQUEST_BYTES) throw invalid("DIAGNOSTIC_REQUEST_TOO_LARGE"); + String key = hash(encode(List.of(order.sourceAccount(), order.warehouseCode(), order.ownerCode(), order.orderNo()))); + String payloadHash = hash(payload); + String configuration = hash(encode(List.of("v1", properties.diagnosticSessionId(), properties.collectionSessionId(), + properties.rootTaskId(), properties.sourceAccount(), properties.warehouseCodes().stream().sorted().distinct().toList(), + properties.startsAt(), properties.expiresAt(), MAX_ORDERS, MAX_ROWS, MAX_BYTES))); + return tx.execute(status -> persist(order, key, payload, payloadHash, bytes, configuration)); + } + private Receipt persist(Order order, String key, String payload, String payloadHash, int bytes, String configuration) { + String id = order.diagnosticSessionId(); + // Duplicate insert acquires the same durable session lock, including first-request races. + jdbc.update(""" + INSERT INTO ll_selfop_quantity_diagnostic_session + (diagnostic_session_id, configuration_hash, source_account, collection_session_id, root_task_id, + starts_at, expires_at, accepted_orders, accepted_sku_rows, accepted_bytes) + VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0, 0) + ON DUPLICATE KEY UPDATE diagnostic_session_id = diagnostic_session_id + """, id, configuration, order.sourceAccount(), order.collectionSessionId(), order.rootTaskId(), + properties.startsAt().toString(), properties.expiresAt().toString()); + Session session = jdbc.queryForObject(""" + SELECT configuration_hash, accepted_orders, accepted_sku_rows, accepted_bytes, stop_reason + FROM ll_selfop_quantity_diagnostic_session WHERE diagnostic_session_id = ? FOR UPDATE + """, (rs, n) -> new Session(rs.getString(1), rs.getInt(2), rs.getInt(3), rs.getLong(4), rs.getString(5)), id); + if (session == null || !configuration.equals(session.configurationHash())) + throw new ResponseStatusException(HttpStatus.CONFLICT, "DIAGNOSTIC_SESSION_CONFIGURATION_CHANGED"); + var existing = jdbc.query(""" + SELECT payload_hash, comparison FROM ll_selfop_quantity_diagnostic_order + WHERE diagnostic_session_id = ? AND order_key_hash = ? + """, (rs, n) -> new Existing(rs.getString(1), Comparison.valueOf(rs.getString(2))), id, key); + if (!existing.isEmpty()) { + if (!payloadHash.equals(existing.get(0).payloadHash())) + throw new ResponseStatusException(HttpStatus.CONFLICT, "DIAGNOSTIC_ORDER_CONFLICT"); + return session.receipt("DUPLICATE", existing.get(0).comparison(), session.stopReason()); + } + if (session.stopReason() != null) return session.receipt("STOPPED", null, session.stopReason()); + Instant now = clock.instant(); + if (now.isBefore(properties.startsAt())) return session.receipt("STOPPED", null, "NOT_STARTED"); + if (!now.isBefore(properties.expiresAt())) return stop(id, session, "EXPIRED"); + if (session.orders() >= MAX_ORDERS || session.rows() + order.skus().size() > MAX_ROWS + || session.bytes() + bytes > MAX_BYTES) return stop(id, session, "LIMIT_REACHED"); + Comparison comparison = compare(order); + jdbc.update(""" + INSERT INTO ll_selfop_quantity_diagnostic_order + (diagnostic_session_id, order_key_hash, source_account, warehouse_code, owner_code, order_no, + payload_hash, comparison, sku_rows, payload_bytes, evidence_json, received_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, id, key, order.sourceAccount(), order.warehouseCode(), order.ownerCode(), order.orderNo(), + payloadHash, comparison.name(), order.skus().size(), bytes, payload, now.toString()); + int orders = session.orders() + 1, rows = session.rows() + order.skus().size(); + long totalBytes = session.bytes() + bytes; + String reason = forecastDifference(order) ? "FORECAST_ACTUAL_DIFFERENCE" + : orders == MAX_ORDERS || rows == MAX_ROWS || totalBytes == MAX_BYTES ? "LIMIT_REACHED" : null; + int updated = jdbc.update(""" + UPDATE ll_selfop_quantity_diagnostic_session + SET accepted_orders = ?, accepted_sku_rows = ?, accepted_bytes = ?, stop_reason = ? + WHERE diagnostic_session_id = ? + """, orders, rows, totalBytes, reason, id); + if (updated != 1) throw new IllegalStateException("DIAGNOSTIC_SESSION_UPDATE_FAILED"); + return new Receipt("ACCEPTED", comparison, reason, orders, rows, totalBytes); + } + private Receipt stop(String id, Session session, String reason) { + jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET stop_reason = ? WHERE diagnostic_session_id = ?", reason, id); + return session.receipt("STOPPED", null, reason); + } + private String encode(Object value) { + try { return json.writeValueAsString(value); } + catch (JsonProcessingException e) { throw invalid("DIAGNOSTIC_ENCODING_FAILED"); } + } + private static String hash(String value) { + try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8))); } + catch (NoSuchAlgorithmException e) { throw new IllegalStateException(e); } + } + private record Existing(String payloadHash, Comparison comparison) { } + private record Session(String configurationHash, int orders, int rows, long bytes, String stopReason) { + Receipt receipt(String outcome, Comparison comparison, String reason) { + return new Receipt(outcome, comparison, reason, orders, rows, bytes); + } + } +} diff --git a/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql b/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql new file mode 100644 index 000000000..eec9e8ae3 --- /dev/null +++ b/lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql @@ -0,0 +1,38 @@ +-- Independent diagnostic tables in the Lastleg schema. No business fact/queue writes. +-- Run only after explicit TEST SQL approval. Not executed by service startup. +CREATE TABLE ll_selfop_quantity_diagnostic_session ( + diagnostic_session_id VARCHAR(128) NOT NULL COMMENT '固定有界数量诊断会话标识', + configuration_hash CHAR(64) NOT NULL COMMENT '冻结范围、UTC起止时间及限额的SHA256摘要', + source_account VARCHAR(64) NOT NULL COMMENT '来源账号别名，不保存凭据', + collection_session_id VARCHAR(128) NOT NULL COMMENT '被观察的来源采集会话标识', + root_task_id VARCHAR(128) NOT NULL COMMENT '被观察的账号采集根任务标识', + starts_at VARCHAR(40) NOT NULL COMMENT '诊断开始时刻，ISO8601 UTC字符串', + expires_at VARCHAR(40) NOT NULL COMMENT '诊断截止时刻，距开始最多30分钟，ISO8601 UTC字符串', + accepted_orders INT NOT NULL DEFAULT 0 COMMENT '已接收完整配对订单数，跨重启累计', + accepted_sku_rows INT NOT NULL DEFAULT 0 COMMENT '已接收归并SKU行数，跨重启累计', + accepted_bytes BIGINT NOT NULL DEFAULT 0 COMMENT '已持久化白名单规范JSON的UTF8字节数', + stop_reason VARCHAR(64) NULL COMMENT '诊断终止原因，空值表示尚未终止', + PRIMARY KEY (diagnostic_session_id), + CONSTRAINT chk_selfop_qdiag_limits CHECK (accepted_orders BETWEEN 0 AND 100 + AND accepted_sku_rows BETWEEN 0 AND 2000 AND accepted_bytes BETWEEN 0 AND 2097152) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='独立数量对照诊断会话，重启不重置限额'; + +CREATE TABLE ll_selfop_quantity_diagnostic_order ( + diagnostic_session_id VARCHAR(128) NOT NULL COMMENT '所属固定诊断会话标识', + order_key_hash CHAR(64) NOT NULL COMMENT '账号、仓库、货主、出库单业务键的SHA256摘要', + source_account VARCHAR(64) NOT NULL COMMENT '来源账号别名', + warehouse_code VARCHAR(64) NOT NULL COMMENT '来源仓库编码', + owner_code VARCHAR(128) NOT NULL COMMENT '来源货主编码', + order_no VARCHAR(128) NOT NULL COMMENT '来源出库单号', + payload_hash CHAR(64) NOT NULL COMMENT '白名单规范载荷SHA256摘要，用于幂等校验', + comparison VARCHAR(40) NOT NULL COMMENT 'SKU集合与三类数量比较结果', + sku_rows INT NOT NULL COMMENT '本单归并后的SKU并集行数', + payload_bytes INT NOT NULL COMMENT '本单白名单规范JSON的UTF8字节数', + evidence_json JSON NOT NULL COMMENT '类型化列表与详情观测及SKU三量证据，不含原始响应或收件信息', + received_at VARCHAR(40) NOT NULL COMMENT '接收时刻，ISO8601 UTC字符串', + PRIMARY KEY (diagnostic_session_id, order_key_hash), + CONSTRAINT fk_selfop_qdiag_session FOREIGN KEY (diagnostic_session_id) + REFERENCES ll_selfop_quantity_diagnostic_session(diagnostic_session_id), + CONSTRAINT chk_selfop_qdiag_order_limits CHECK (sku_rows BETWEEN 1 AND 200 + AND payload_bytes BETWEEN 1 AND 262144) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='独立配对订单数量诊断证据，不参与销量发布'; diff --git a/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml b/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml new file mode 100644 index 000000000..63708501c --- /dev/null +++ b/lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml @@ -0,0 +1,105 @@ +openapi: 3.0.3 +info: + title: Internal bounded self-operated quantity diagnostics + version: 1.0.0 +paths: + /internal/lastleg/self-operated-analytics/quantity-diagnostics/orders: + post: + operationId: receiveSelfopQuantityDiagnosticOrder + description: >- + Internal service-token only; never routed through Gateway. One complete paired order per request. + Quantities are evidence only. Session scope and UTC time interval are frozen by server configuration. + Maximum 100 orders, 2000 SKU rows and 2097152 canonical UTF8 bytes per session; + 200 SKU rows and 262144 wire/canonical bytes per request; at most 30 minutes. + First known forecast/actual difference closes capture regardless of comparison classification. + Same identity and canonical evidence is idempotent, including after closure; changed evidence is 409. + security: + - internalServiceToken: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/PairedOrder' + responses: + '200': + description: Diagnostic receipt in the standard ApiResponse wrapper; code is string 200. + content: + application/json: + schema: + type: object + properties: + code: { type: string, example: '200' } + data: { $ref: '#/components/schemas/Receipt' } + '400': { description: 'Invalid JSON, unknown fields, invalid scope/evidence, duplicate SKU or request limit.' } + '401': { description: Invalid internal service token. } + '409': { description: Session configuration changed or order identity reused with different evidence. } + '503': { description: Internal service token unconfigured or diagnostic storage unavailable. } +components: + securitySchemes: + internalServiceToken: + type: apiKey + in: header + name: X-Internal-Service-Token + schemas: + PairedOrder: + type: object + additionalProperties: false + required: [diagnosticSessionId, collectionSessionId, rootTaskId, sourceAccount, warehouseCode, ownerCode, orderNo, listObservation, detailObservation, skus] + properties: + diagnosticSessionId: { type: string, maxLength: 128 } + collectionSessionId: { type: string, maxLength: 128 } + rootTaskId: { type: string, maxLength: 128 } + sourceAccount: { type: string, maxLength: 64 } + warehouseCode: { type: string, maxLength: 64 } + ownerCode: { type: string, maxLength: 128 } + orderNo: { type: string, maxLength: 128 } + listObservation: { $ref: '#/components/schemas/Observation' } + detailObservation: { $ref: '#/components/schemas/Observation' } + skus: + type: array + minItems: 1 + maxItems: 200 + description: Union of list/detail SKUs; duplicate source lines aggregated, unknown never zero-filled. + items: { $ref: '#/components/schemas/Sku' } + Observation: + type: object + additionalProperties: false + required: [observedAt, attemptId, skuSetComplete] + properties: + status: { type: string, nullable: true, maxLength: 32 } + outboundTime: + type: string + nullable: true + maxLength: 32 + description: Original Asia/Shanghai source text, null if absent; not reinterpreted as another timezone. + observedAt: { type: string, format: date-time } + attemptId: { type: string, maxLength: 128 } + skuSetComplete: { type: boolean } + Sku: + type: object + additionalProperties: false + required: [sku, listPresent, detailPresent] + properties: + sku: { type: string, maxLength: 128 } + listQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 } + forecastQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 } + actualQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 } + listPresent: { type: boolean } + detailPresent: { type: boolean } + Receipt: + type: object + required: [outcome, acceptedOrders, acceptedSkuRows, acceptedBytes] + properties: + outcome: { type: string, enum: [ACCEPTED, DUPLICATE, STOPPED] } + comparison: + type: string + nullable: true + enum: [SKU_SET_MISMATCH, QUANTITY_UNKNOWN, FORECAST_ACTUAL_DIFFERENCE, ALL_EQUAL, LIST_QUANTITY_DIFFERENCE] + stopReason: + type: string + nullable: true + enum: [DISABLED, NOT_STARTED, EXPIRED, LIMIT_REACHED, FORECAST_ACTUAL_DIFFERENCE] + acceptedOrders: { type: integer } + acceptedSkuRows: { type: integer } + acceptedBytes: { type: integer, format: int64, description: Canonical whitelist JSON UTF8 bytes persisted in this session. } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java new file mode 100644 index 000000000..3e36b1723 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticControllerTest.java @@ -0,0 +1,47 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import com.baiyue.lastleg.security.InternalServiceTokenGuard; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; + +class QuantityDiagnosticControllerTest { + @Test void httpStatusContractSurvivesSharedBusinessErrorAdvice() throws Exception { + var service = mock(QuantityDiagnosticService.class); + var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"), service, new ObjectMapper()); + var mvc = org.springframework.test.web.servlet.setup.MockMvcBuilders.standaloneSetup(controller) + .setControllerAdvice(new com.baiyue.common.web.GlobalExceptionHandler()).build(); + String path = "/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders"; + mvc.perform(org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post(path) + .contentType("application/json").content("{}")) + .andExpect(org.springframework.test.web.servlet.result.MockMvcResultMatchers.status().isUnauthorized()); + mvc.perform(org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post(path) + .header("X-Internal-Service-Token", "test-token").contentType("application/json").content("{\"address\":1}")) + .andExpect(org.springframework.test.web.servlet.result.MockMvcResultMatchers.status().isBadRequest()); + } + @Test void rejectsInvalidTokenBeforeParsingOrStorage() { + var service = mock(QuantityDiagnosticService.class); + var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"), service, new ObjectMapper()); + assertThatThrownBy(() -> controller.accept(null, new byte[0])).hasMessageContaining("401"); + assertThatThrownBy(() -> controller.accept("wrong", new byte[0])).hasMessageContaining("401"); + verifyNoInteractions(service); + } + @Test void missingTokenConfigurationFailsClosed() { + var service = mock(QuantityDiagnosticService.class); + var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard(""), service, new ObjectMapper()); + assertThatThrownBy(() -> controller.accept("any", new byte[0])).hasMessageContaining("503"); + verifyNoInteractions(service); + } + @Test void rejectsNonWhitelistedFieldsTrailingDocumentsAndOversizedPayload() { + var service = mock(QuantityDiagnosticService.class); + var controller = new QuantityDiagnosticController(new InternalServiceTokenGuard("test-token"), service, new ObjectMapper()); + assertThatThrownBy(() -> controller.accept("test-token", "{\"address\":\"not allowed\"}".getBytes(StandardCharsets.UTF_8))) + .hasMessageContaining("DIAGNOSTIC_JSON_INVALID"); + assertThatThrownBy(() -> controller.accept("test-token", "{} {}".getBytes(StandardCharsets.UTF_8))) + .hasMessageContaining("DIAGNOSTIC_JSON_INVALID"); + assertThatThrownBy(() -> controller.accept("test-token", new byte[262145])).hasMessageContaining("DIAGNOSTIC_REQUEST_TOO_LARGE"); + verifyNoInteractions(service); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java new file mode 100644 index 000000000..d0ab0ab8b --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticServiceTest.java @@ -0,0 +1,169 @@ +package com.baiyue.lastleg.selfoperated.analytics.diagnostic; + +import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*; +import static org.assertj.core.api.Assertions.*; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.time.*; +import java.util.*; +import java.util.concurrent.*; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.core.io.ClassPathResource; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import org.springframework.web.server.ResponseStatusException; + +class QuantityDiagnosticServiceTest { + private final Instant start = Instant.parse("2026-10-13T15:00:00Z"); + private final ObjectMapper json = new ObjectMapper().findAndRegisterModules(); + private JdbcTemplate jdbc; + private DataSourceTransactionManager manager; + private QuantityDiagnosticService service; + private QuantityDiagnosticProperties properties; + + @BeforeEach void prepare() throws Exception { + var ds = new DriverManagerDataSource("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=2000", "sa", ""); + jdbc = new JdbcTemplate(ds); manager = new DataSourceTransactionManager(ds); + // Exercise the dedicated production DDL, omitting only MySQL table options/comments. + String ddl; + try (var in = new ClassPathResource("db/selfop-quantity-diagnostics.sql").getInputStream()) { + ddl = new String(in.readAllBytes(), StandardCharsets.UTF_8); + } + ddl = ddl.replaceAll("(?m)^--.*$", "").replaceAll("COMMENT\\s*=?\\s*'[^']*'", "") + .replaceAll("ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin", ""); + for (String statement : ddl.split(";")) if (!statement.isBlank()) jdbc.execute(statement); + properties = config(start.plusSeconds(1800), true); + service = instance(properties, start.plusSeconds(10)); + } + private QuantityDiagnosticProperties config(Instant expires, boolean enabled) { + return new QuantityDiagnosticProperties(enabled, "diag-1", "collection-1", "root-1", "account-1", + List.of("warehouse-1"), start, expires); + } + private QuantityDiagnosticService instance(QuantityDiagnosticProperties p, Instant now) { + return new QuantityDiagnosticService(jdbc, manager, json, p, Clock.fixed(now, ZoneOffset.UTC)); + } + private Sku sku(String code, String list, String forecast, String actual) { + return new Sku(code, list == null ? null : new BigDecimal(list), forecast == null ? null : new BigDecimal(forecast), + actual == null ? null : new BigDecimal(actual), true, true); + } + private Order order(String no, Sku... skus) { + return new Order("diag-1", "collection-1", "root-1", "account-1", "warehouse-1", "owner-1", no, + new Observation("100", "2026-10-06 12:00:00", start.plusSeconds(1), "list-1", true), + new Observation("100", "2026-10-06 12:00:00", start.plusSeconds(2), "detail-1", true), List.of(skus)); + } + @Test void firstDifferencePersistsEvidenceAndStopsFurtherOrdersIncludingAfterRestart() { + var request = order("order-1", sku("sku-1", "10", "10", "8")); + var result = service.accept(request); + assertThat(result.outcome()).isEqualTo("ACCEPTED"); + assertThat(result.comparison()).isEqualTo(Comparison.FORECAST_ACTUAL_DIFFERENCE); + assertThat(result.stopReason()).isEqualTo("FORECAST_ACTUAL_DIFFERENCE"); + service = instance(properties, start.plusSeconds(20)); + assertThat(service.accept(request).outcome()).isEqualTo("DUPLICATE"); + assertThat(service.accept(order("order-2", sku("sku-1", "1", "1", "1"))).outcome()).isEqualTo("STOPPED"); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order", Integer.class)).isEqualTo(1); + } + @Test void listMatchingActualStillRecordsForecastDifference() { + assertThat(service.accept(order("order-1", sku("sku", "8", "10", "8"))).comparison()) + .isEqualTo(Comparison.FORECAST_ACTUAL_DIFFERENCE); + } + @Test void idempotencyCanonicalizesDecimalsAndRowOrderButRejectsChangedEvidence() { + service.accept(order("order-1", sku("b", "1.0", "1", "1"), sku("a", "2", "2", "2"))); + var result = service.accept(order("order-1", sku("a", "2.00", "2", "2"), sku("b", "1", "1", "1"))); + assertThat(result.outcome()).isEqualTo("DUPLICATE"); + assertThat(result.acceptedOrders()).isEqualTo(1); + assertThatThrownBy(() -> service.accept(order("order-1", sku("a", "3", "3", "3")))) + .isInstanceOf(ResponseStatusException.class).hasMessageContaining("409"); + } + @Test void skuReplacementCannotBeHiddenByEqualTotalsAndDifferenceStillStops() { + var result = service.accept(order("order-1", + new Sku("old", BigDecimal.TEN, null, null, true, false), + new Sku("new", null, BigDecimal.TEN, BigDecimal.valueOf(8), false, true))); + assertThat(result.comparison()).isEqualTo(Comparison.SKU_SET_MISMATCH); + assertThat(result.stopReason()).isEqualTo("FORECAST_ACTUAL_DIFFERENCE"); + } + @Test void missingQuantityRemainsUnknown() { + var result = service.accept(order("order-1", sku("sku", null, "1", "1"))); + assertThat(result.comparison()).isEqualTo(Comparison.QUANTITY_UNKNOWN); + assertThat(result.stopReason()).isNull(); + } + @Test void duplicateSkusAndCrossGenerationAreRejectedWithoutWrites() { + assertThatThrownBy(() -> service.accept(order("order-1", sku("sku", "1", "1", "1"), sku("sku", "1", "1", "1")))) + .hasMessageContaining("DIAGNOSTIC_DUPLICATE_SKU"); + Order r = order("order-1", sku("sku", "1", "1", "1")); + assertThatThrownBy(() -> service.accept(new Order(r.diagnosticSessionId(), "other-generation", r.rootTaskId(), r.sourceAccount(), + r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), r.detailObservation(), r.skus()))) + .hasMessageContaining("DIAGNOSTIC_SCOPE_MISMATCH"); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_session", Integer.class)).isZero(); + } + @Test void expiredAndReconfiguredSessionsCannotResetCounters() { + var r = order("order-1", sku("sku", "1", "1", "1")); service.accept(r); + var expired = instance(properties, start.plusSeconds(1800)); + assertThat(expired.accept(order("order-2", sku("sku", "1", "1", "1"))).stopReason()).isEqualTo("EXPIRED"); + assertThat(expired.accept(r).outcome()).isEqualTo("DUPLICATE"); + assertThatThrownBy(() -> instance(config(start.plusSeconds(1799), true), start.plusSeconds(20)).accept(r)) + .hasMessageContaining("DIAGNOSTIC_SESSION_CONFIGURATION_CHANGED"); + } + @Test void disabledRequiresNoTablesOrValidSession() { + jdbc.execute("DROP TABLE ll_selfop_quantity_diagnostic_order"); + jdbc.execute("DROP TABLE ll_selfop_quantity_diagnostic_session"); + assertThat(instance(config(start.plusSeconds(1800), false), start).accept(null).stopReason()).isEqualTo("DISABLED"); + } + @Test void concurrentRequestsCannotExceedHundredOrders() throws Exception { + var pool = Executors.newFixedThreadPool(4); + try { + List> jobs = new ArrayList<>(); + for (int i = 0; i < 104; i++) { final int n = i; jobs.add(() -> service.accept(order("order-" + n, sku("sku", "1", "1", "1")))); } + var results = pool.invokeAll(jobs); + int accepted = 0; + for (var result : results) if (result.get().outcome().equals("ACCEPTED")) accepted++; + assertThat(accepted).isEqualTo(100); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order", Integer.class)).isEqualTo(100); + assertThat(instance(properties, start.plusSeconds(20)).accept(order("after-restart", sku("sku", "1", "1", "1"))).stopReason()) + .isEqualTo("LIMIT_REACHED"); + } finally { pool.shutdownNow(); } + } + @Test void rowLimitAndFailedOrderInsertDoNotPartiallyAdvanceCounters() { + service.accept(order("order-1", sku("sku", "1", "1", "1"))); + jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET accepted_sku_rows=2000"); + assertThat(service.accept(order("order-2", sku("sku", "1", "1", "1"))).stopReason()).isEqualTo("LIMIT_REACHED"); + assertThat(jdbc.queryForObject("SELECT accepted_orders FROM ll_selfop_quantity_diagnostic_session", Integer.class)).isEqualTo(1); + } + @Test void byteLimitCannotBeResetByRestartOrPartiallySaveOrder() { + service.accept(order("order-1", sku("sku", "1", "1", "1"))); + jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET accepted_bytes=2097150"); + var receipt = instance(properties, start.plusSeconds(20)).accept(order("order-2", sku("sku", "1", "1", "1"))); + assertThat(receipt.stopReason()).isEqualTo("LIMIT_REACHED"); + assertThat(receipt.acceptedBytes()).isEqualTo(2097150); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_order", Integer.class)).isEqualTo(1); + } + @Test void dedicatedDdlHasChineseCommentsOnEveryTableAndOrdinaryColumn() throws Exception { + String ddl; + try (var in = new ClassPathResource("db/selfop-quantity-diagnostics.sql").getInputStream()) { + ddl = new String(in.readAllBytes(), StandardCharsets.UTF_8); + } + var fields = java.util.regex.Pattern.compile("(?m)^ [a-z][a-z_]+ (?:VARCHAR|CHAR|INT|BIGINT|JSON)\\b[^\\n]*").matcher(ddl); + int count = 0; + while (fields.find()) { + assertThat(fields.group()).matches("(?s).*COMMENT '[^']*[\\u4e00-\\u9fff][^']*'.*"); count++; + } + assertThat(count).isEqualTo(23); + assertThat(java.util.regex.Pattern.compile("COMMENT='[^']*[\\u4e00-\\u9fff][^']*';").matcher(ddl).results().count()).isEqualTo(2); + } + @Test void storageFailureRollsBackSessionCounterAndSessionCreation() { + jdbc.execute("ALTER TABLE ll_selfop_quantity_diagnostic_order ADD CONSTRAINT reject_test_order CHECK(order_no <> 'bad-order')"); + assertThatThrownBy(() -> service.accept(order("bad-order", sku("sku", "1", "1", "1")))) + .isInstanceOf(org.springframework.dao.DataIntegrityViolationException.class); + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_quantity_diagnostic_session", Integer.class)).isZero(); + } + @Test void cancellationAndMissingOutboundTimeRemainEvidenceOnly() { + var r = order("cancelled", sku("sku", "1", "1", "1")); + var cancelled = new Observation("cancelled", null, start.plusSeconds(2), "detail-cancel", true); + var result = service.accept(new Order(r.diagnosticSessionId(), r.collectionSessionId(), r.rootTaskId(), r.sourceAccount(), + r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), cancelled, r.skus())); + assertThat(result.comparison()).isEqualTo(Comparison.ALL_EQUAL); + assertThat(result.acceptedOrders()).isEqualTo(1); + } +}