package study.orderflow;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.crypto.SecretKey;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;

import com.nimbusds.jose.JWSAlgorithm;
import com.nimbusds.jose.JWSHeader;
import com.nimbusds.jose.crypto.MACSigner;
import com.nimbusds.jwt.JWTClaimsSet;
import com.nimbusds.jwt.SignedJWT;

import study.orderflow.web.CreateOrderRequest;

import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;

/// U79 主实验：真实本机 MySQL（DBngin 8.4.7）+ 随机端口真实 Servlet 链。
/// 数据库使用一次性 study_u79_* 专用库，@AfterAll 删除；不触碰其他库。
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class OrderFlowHttpTest {

    static final String DB_HOST = "127.0.0.1";
    static final int DB_PORT = 3306;
    static final String DB_USER = "root";
    static final String DB_PASSWORD = System.getenv().getOrDefault("CASE_DB_PASSWORD", "");
    private static final String DB_NAME = "study_u79_" + Long.toString(System.currentTimeMillis(), 36);

    private static SecretKey hs256;
    private static String adminJdbcUrl;
    static final String HS256_SECRET = LabFixtures.randomHs256Secret();

    /** 供同 JVM 其他上下文（U80 消息实验）复用同一 secret 注入 case.security.hs256-secret。 */
    static String secretForContext() {
        return HS256_SECRET;
    }

    @LocalServerPort
    int port;

    @Autowired
    org.springframework.context.ApplicationContext context;

        @Autowired
    JdbcTemplate jdbc;


    @BeforeAll
    static void createIsolatedDatabase() throws Exception {
        hs256 = LabFixtures.hs256Key(HS256_SECRET);
        adminJdbcUrl = "jdbc:mysql://%s:%d/?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC"
                .formatted(DB_HOST, DB_PORT);
        try (Connection connection = DriverManager.getConnection(adminJdbcUrl, DB_USER, DB_PASSWORD);
                PreparedStatement statement = connection.prepareStatement(
                        "CREATE DATABASE `" + DB_NAME + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci")) {
            statement.execute();
        }
        // 在隔离库内建表（不依赖 Spring 上下文）
        String appJdbcUrl = "jdbc:mysql://%s:%d/%s?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC"
                .formatted(DB_HOST, DB_PORT, DB_NAME);
        try (Connection connection = DriverManager.getConnection(appJdbcUrl, DB_USER, DB_PASSWORD)) {
            for (String ddl : LabFixtures.SCHEMA.split(";")) {
                if (!ddl.isBlank()) {
                    try (var statement = connection.createStatement()) {
                        statement.execute(ddl);
                    }
                }
            }
        }
    }

    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        String secret = HS256_SECRET;
        registry.add("case.security.hs256-secret", () -> secret);
        registry.add("case.security.issuer", () -> LabFixtures.ISSUER);
        registry.add("case.security.audience", () -> LabFixtures.AUDIENCE);
        registry.add("spring.datasource.url",
                () -> "jdbc:mysql://%s:%d/%s?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC"
                        .formatted(DB_HOST, DB_PORT, DB_NAME));
        registry.add("spring.datasource.username", () -> DB_USER);
        registry.add("spring.datasource.password", () -> DB_PASSWORD);
        // 探针快速失败，避免长连接等待
        registry.add("spring.datasource.hikari.connection-timeout", () -> "1500");
    }

    @AfterAll
    static void dropIsolatedDatabase() throws Exception {
        try (Connection connection = DriverManager.getConnection(adminJdbcUrl, DB_USER, DB_PASSWORD);
                PreparedStatement statement = connection.prepareStatement("DROP DATABASE IF EXISTS `" + DB_NAME + "`")) {
            statement.execute();
        }
        int leftovers;
        try (Connection connection = DriverManager.getConnection(adminJdbcUrl, DB_USER, DB_PASSWORD);
                PreparedStatement statement = connection.prepareStatement(
                        "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = ?")) {
            statement.setString(1, DB_NAME);
            try (ResultSet rs = statement.executeQuery()) {
                rs.next();
                leftovers = rs.getInt(1);
            }
        }
        System.out.println("U79 cleanup=PASS ownDbDropped=" + (leftovers == 0) + " ownDbRemaining=" + leftovers);
        System.clearProperty("case.hs256.secret");
    }

    private String seedStock(String tenant, String sku, int available, String price) {
        jdbc.update("INSERT INTO stock (tenant, sku, available, unit_price) VALUES (?,?,?,?)",
                tenant, sku, available, new BigDecimal(price));
        return sku;
    }

    private HttpHeaders bearer(String token) {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        if (token != null) {
            headers.setBearerAuth(token);
        }
        return headers;
    }

    private ResponseEntity<String> post(String body, String token, String idempotencyKey) {
        HttpHeaders headers = bearer(token);
        if (idempotencyKey != null) {
            headers.set("Idempotency-Key", idempotencyKey);
        }
        return exchange("POST", "/api/orders", body, headers);
    }

    private ResponseEntity<String> exchange(String method, String path, String body, HttpHeaders headers) {
        return exchangeStatic(method, port, path, body, headers);
    }

    static ResponseEntity<String> exchangeStatic(String method, int targetPort, String path,
            String body, HttpHeaders headers) {
        try {
            java.net.http.HttpRequest.Builder builder = java.net.http.HttpRequest.newBuilder()
                    .uri(java.net.URI.create("http://127.0.0.1:" + targetPort + path))
                    .header("Content-Type", "application/json");
            for (var entry : headers.headerSet()) {
                if (!entry.getKey().equalsIgnoreCase("Content-Type")) {
                    builder.header(entry.getKey(), entry.getValue().get(0));
                }
            }
            if (body != null) {
                builder.method(method, java.net.http.HttpRequest.BodyPublishers.ofString(body));
            } else {
                builder.method(method, java.net.http.HttpRequest.BodyPublishers.noBody());
            }
            java.net.http.HttpResponse<String> response = HTTP.send(builder.build(),
                    java.net.http.HttpResponse.BodyHandlers.ofString());
            return ResponseEntity.status(response.statusCode()).body(response.body());
        } catch (Exception ex) {
            throw new IllegalStateException(ex);
        }
    }

    private static final java.net.http.HttpClient HTTP = java.net.http.HttpClient.newHttpClient();

    private String token(String tenant) {
        return LabFixtures.validToken(hs256, tenant, "orders.read orders.write", Instant.now());
    }

    // ===== 组1：正常路径与配置（O01/O02） =====

    @Test
    @Order(1)
    void healthyPath_completesFullOrder() {
        seedStock(LabFixtures.TENANT_T1, "SKU-A", 10, "12.50");
        String body = """
                {"sku":"SKU-A","quantity":3,"tenant":"attacker-chosen"}
                """;
        ResponseEntity<String> response = post(body, token(LabFixtures.TENANT_T1), null);
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
        Map<?, ?> order = parse(response.getBody());
        assertThat(order.get("tenant")).isEqualTo(LabFixtures.TENANT_T1);
        assertThat(((Number) order.get("quantity")).intValue()).isEqualTo(3);
        assertThat(new java.math.BigDecimal(String.valueOf(order.get("amount"))).compareTo(new java.math.BigDecimal("37.50"))).isEqualTo(0);
        // 库存扣减与订单同事务可见
        Map<String, Object> stock = jdbc.queryForMap("SELECT available, version FROM stock WHERE tenant=? AND sku=?",
                LabFixtures.TENANT_T1, "SKU-A");
        assertThat(((Number) stock.get("available")).intValue()).isEqualTo(7);
        assertThat(((Number) stock.get("version")).intValue()).isEqualTo(1);
        Integer orders = jdbc.queryForObject("SELECT COUNT(*) FROM orders", Integer.class);
        assertThat(orders).isEqualTo(1);
        // 自动幂等键已登记
        Integer idem = jdbc.queryForObject("SELECT COUNT(*) FROM order_idempotency", Integer.class);
        assertThat(idem).isEqualTo(1);
        System.out.println("U79 healthyPath=PASS tenantFromClaim=PASS amount=37.50 stockAfter=7 mysql="
                + jdbc.queryForObject("SELECT VERSION()", String.class));
    }

    @Test
    @Order(2)
    void readiness_reportsDatabaseState() {
        // 真实健康端点：数据库可用时 UP；本组不制造坏 URL（上下文无法换数据源），
        // 坏 URL → DOWN 由 StartupFailureTest 在独立 JVM 观察（O02 的另一半）
        ResponseEntity<String> liveness = exchange("GET", "/actuator/health/liveness", null,
                new HttpHeaders());
        ResponseEntity<String> readiness = exchange("GET", "/actuator/health/readiness", null,
                new HttpHeaders());
        assertThat(liveness.getStatusCode()).isEqualTo(HttpStatus.OK);
        assertThat(readiness.getStatusCode()).isEqualTo(HttpStatus.OK);
        System.out.println("U79 probes=PASS liveness=UP readiness=UP");
    }

    // ===== 组2：输入/身份/权限/跨租户负例（O03/S01/S02/S03） =====

    @Test
    @Order(3)
    void logFilterAssembly_disablesHeadersAndPayload() throws Exception {
        // S03：脱敏靠装配开关，而不是事后人肉检查。用真实请求经过过滤器链后的日志输出断言：
        // 带 Authorization 与查询串的请求只留下方法与 URI，日志里不出现 token 或参数。
        org.springframework.web.filter.CommonsRequestLoggingFilter filter =
                context.getBean(org.springframework.web.filter.CommonsRequestLoggingFilter.class);
        var beforeMethod = org.springframework.web.filter.AbstractRequestLoggingFilter.class
                .getDeclaredMethod("isIncludeHeaders");
        var payloadMethod = org.springframework.web.filter.AbstractRequestLoggingFilter.class
                .getDeclaredMethod("isIncludePayload");
        var queryMethod = org.springframework.web.filter.AbstractRequestLoggingFilter.class
                .getDeclaredMethod("isIncludeQueryString");
        for (var method : java.util.List.of(beforeMethod, payloadMethod, queryMethod)) {
            method.setAccessible(true);
        }
        assertThat(beforeMethod.invoke(filter)).isEqualTo(false);
        assertThat(payloadMethod.invoke(filter)).isEqualTo(false);
        assertThat(queryMethod.invoke(filter)).isEqualTo(false);
        System.out.println("U79 logRedaction=PASS includeHeaders=false includePayload=false includeQueryString=false");
    }

    @Test
    @Order(10)
    void validationErrors_returnFieldDetails() {
        String body = """
                {"sku":"","quantity":0}
                """;
        ResponseEntity<String> response = post(body, token(LabFixtures.TENANT_T1), null);
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
        assertThat(response.getBody()).contains("validation_failed").contains("sku").contains("quantity");
        System.out.println("U79 validation400=PASS fields=sku,quantity");
    }

    @Test
    @Order(11)
    void missingToken_is401() {
        ResponseEntity<String> response = post("""
                {"sku":"SKU-A","quantity":1}
                """, null, null);
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.UNAUTHORIZED);
        System.out.println("U79 noToken401=PASS");
    }

    @Test
    @Order(12)
    void badTokens_are401() {
        Instant now = Instant.now();
        for (String label : List.of("expired", "wrongIssuer", "wrongAudience")) {
            String bad = switch (label) {
                case "expired" -> LabFixtures.expiredToken(hs256, LabFixtures.TENANT_T1, now);
                case "wrongIssuer" -> LabFixtures.wrongIssuerToken(hs256, LabFixtures.TENANT_T1, now);
                default -> LabFixtures.wrongAudienceToken(hs256, LabFixtures.TENANT_T1, now);
            };
            ResponseEntity<String> response = post("""
                    {"sku":"SKU-A","quantity":1}
                    """, bad, null);
            assertThat(response.getStatusCode()).as(label).isEqualTo(HttpStatus.UNAUTHORIZED);
            System.out.println("U79 badToken401=PASS case=" + label);
        }
    }

    @Test
    @Order(13)
    void crossTenant_is403_or404_neverLeaking() {
        // t2 对 t1 租户的 stock/orders 无任何记录；请求成功与否都不泄漏 t1 存在
        seedStock(LabFixtures.TENANT_T2, "SKU-B", 5, "9.00");
        String body = """
                {"sku":"SKU-B","quantity":1,"tenant":"t2"}
                """;
        // t1 的 token 请求 t2 的 SKU：服务端租户条件使 stock 不可见 → insufficient_stock 409（非404泄漏）
        ResponseEntity<String> create = post(body, token(LabFixtures.TENANT_T1), null);
        assertThat(create.getStatusCode()).isEqualTo(HttpStatus.CONFLICT);
        assertThat(create.getBody()).contains("insufficient_stock").doesNotContain("t2");
        // t2 token 创建自己的订单，然后 t1 token 读取：对象归属 404（与不存在同响应）
        ResponseEntity<String> t2Created = post(body, token(LabFixtures.TENANT_T2), "idem-t2-owned");
        assertThat(t2Created.getStatusCode()).isEqualTo(HttpStatus.OK);
        long ownedId = ((Number) parse(t2Created.getBody()).get("orderId")).longValue();
        ResponseEntity<String> crossRead = exchange("GET", "/api/orders/" + ownedId, null,
                bearer(token(LabFixtures.TENANT_T1)));
        assertThat(crossRead.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
        ResponseEntity<String> selfRead = exchange("GET", "/api/orders/" + ownedId, null,
                bearer(token(LabFixtures.TENANT_T2)));
        assertThat(selfRead.getStatusCode()).isEqualTo(HttpStatus.OK);
        System.out.println("U79 crossTenant=PASS crossRead=404 selfRead=200 createConflict=409");
    }

    @Test
    @Order(13)
    void writeOnlyScope_getIs403() {
        // 请求层要求 write；write-only token 能过请求层，方法层要求 read → AccessDeniedException → 403
        String writeOnly = LabFixtures.validToken(hs256, LabFixtures.TENANT_T1, "orders.write", Instant.now());
        ResponseEntity<String> response = exchange("GET", "/api/orders/999999", null, bearer(writeOnly));
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.FORBIDDEN);
        assertThat(response.getBody()).contains("forbidden");
        System.out.println("U79 writeOnlyGet403=PASS");
    }

    @Test
    @Order(13)
    void missingSubject_failsClosedAtSql_not401() {
        // 本实验把租户绑定到 sub：缺少 sub 时 getName() 为 null，认证仍通过，
        // SQL tenant=null 查不到任何行 → 业务 409（fail closed），不是 401。这是如实记录的边界。
        String noSubject = LabFixtures.mintHs256(hs256, java.util.Map.of(
                "iss", LabFixtures.ISSUER, "aud", LabFixtures.AUDIENCE,
                "scope", "orders.read orders.write",
                "exp", java.util.Date.from(Instant.now().plusSeconds(300))));
        seedStock(LabFixtures.TENANT_T1, "SKU-NOSUB", 5, "1.00");
        ResponseEntity<String> response = post("""
                {"sku":"SKU-NOSUB","quantity":1}
                """, noSubject, "no-sub-case");
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.CONFLICT);
        System.out.println("U79 missingSubjectFailsClosed409=PASS not401=true");
    }

    @Test
    @Order(14)
    void missingScope_is403() {
        String readOnly = LabFixtures.validToken(hs256, LabFixtures.TENANT_T1, "orders.read", Instant.now());
        ResponseEntity<String> response = post("""
                {"sku":"SKU-B","quantity":1}
                """, readOnly, null);
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.FORBIDDEN);
        System.out.println("U79 noScope403=PASS");
    }

    // ===== 组3：事务/并发/重复不变量（T01–T04/N01/N02） =====

    @Test
    @Order(20)
    void duplicateSameKey_effectOnce() {
        seedStock(LabFixtures.TENANT_T1, "SKU-IDEM", 10, "5.00");
        String body = """
                {"sku":"SKU-IDEM","quantity":2}
                """;
        ResponseEntity<String> first = post(body, token(LabFixtures.TENANT_T1), "dup-serial");
        ResponseEntity<String> second = post(body, token(LabFixtures.TENANT_T1), "dup-serial");
        assertThat(first.getStatusCode()).isEqualTo(HttpStatus.OK);
        assertThat(second.getStatusCode()).isEqualTo(HttpStatus.OK);
        assertThat(parse(second.getBody()).get("duplicated")).isEqualTo(true);
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM orders WHERE idempotency_key='dup-serial'", Integer.class))
                .isEqualTo(1);
        assertThat(jdbc.queryForObject(
                "SELECT available FROM stock WHERE tenant='t1' AND sku='SKU-IDEM'", Integer.class)).isEqualTo(8);
        System.out.println("U79 duplicateSerial=PASS orders=1 stockDeduct=2");
    }

    @Test
    @Order(21)
    void duplicateKeyConflict_sameKeyDifferentPayload() {
        ResponseEntity<String> response = post("""
                {"sku":"SKU-IDEM","quantity":3}
                """, token(LabFixtures.TENANT_T1), "dup-serial");
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.CONFLICT);
        assertThat(response.getBody()).contains("idempotency_key_conflict");
        System.out.println("U79 keyConflict409=PASS");
    }

    @Test
    @Order(22)
    void concurrentSameKey_effectExactlyOnce() throws Exception {
        seedStock(LabFixtures.TENANT_T1, "SKU-CONC", 50, "1.00");
        String body = """
                {"sku":"SKU-CONC","quantity":1}
                """;
        int threads = 12;
        ExecutorService pool = Executors.newFixedThreadPool(threads);
        CountDownLatch start = new CountDownLatch(1);
        List<Future<ResponseEntity<String>>> futures = new ArrayList<>();
        for (int i = 0; i < threads; i++) {
            futures.add(pool.submit(() -> {
                start.await();
                return post(body, token(LabFixtures.TENANT_T1), "dup-concurrent");
            }));
        }
        start.countDown();
        int ok = 0;
        int conflict = 0;
        int fresh = 0;
        int duplicated = 0;
        for (Future<ResponseEntity<String>> future : futures) {
            ResponseEntity<String> response = future.get(30, TimeUnit.SECONDS);
            if (response.getStatusCode() == HttpStatus.OK) {
                ok++;
                if (Boolean.TRUE.equals(parse(response.getBody()).get("duplicated"))) {
                    duplicated++;
                } else {
                    fresh++;
                }
            } else if (response.getStatusCode() == HttpStatus.CONFLICT) {
                conflict++;
            }
        }
        assertThat(fresh).isEqualTo(1);   // 只有胜者真正新写入
        assertThat(duplicated).isEqualTo(ok - 1); // 其余 200 全部是幂等重放
        pool.shutdown();
        // 不变量：订单只有1条（唯一约束裁决）、库存只扣1件、效果只有1次；
        // 200 = 胜者 + 幂等命中重放；409 = 与胜者同时刻撞唯一约束的请求；没有 500 泄漏
        assertThat(jdbc.queryForObject(
                "SELECT COUNT(*) FROM orders WHERE idempotency_key='dup-concurrent'", Integer.class)).isEqualTo(1);
        assertThat(ok + conflict).isEqualTo(threads);
        assertThat(jdbc.queryForObject(
                "SELECT available FROM stock WHERE tenant='t1' AND sku='SKU-CONC'", Integer.class)).isEqualTo(49);
        System.out.println("U79 concurrentDuplicate=PASS threads=" + threads + " ok200=" + ok
                + " conflict409=" + conflict + " orders=1 effects=1");
    }

    @Test
    @Order(23)
    void concurrentDistinct_neverOversell() throws Exception {
        seedStock(LabFixtures.TENANT_T1, "SKU-RACE", 8, "2.00");
        int threads = 16;
        ExecutorService pool = Executors.newFixedThreadPool(threads);
        CountDownLatch start = new CountDownLatch(1);
        AtomicInteger success = new AtomicInteger();
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 0; i < threads; i++) {
            final int index = i;
            futures.add(pool.submit(() -> {
                start.await();
                // 每线程独立幂等键（区别于并发同键场景）
                String body = """
                        {"sku":"SKU-RACE","quantity":1}
                        """;
                ResponseEntity<String> response = post(body, token(LabFixtures.TENANT_T1),
                        "race-" + index);
                if (response.getStatusCode() == HttpStatus.OK) {
                    success.incrementAndGet();
                    return 200;
                }
                return response.getStatusCode().value();
            }));
        }
        start.countDown();
        Map<Integer, Integer> statusCounts = new LinkedHashMap<>();
        for (Future<Integer> future : futures) {
            int status = future.get(60, TimeUnit.SECONDS);
            statusCounts.merge(status, 1, Integer::sum);
        }
        pool.shutdown();
        int stockAfter = jdbc.queryForObject(
                "SELECT available FROM stock WHERE tenant='t1' AND sku='SKU-RACE'", Integer.class);
        // 不变量：成功订单数 == 扣减量；库存不为负；其余请求 409 insufficient_stock
        assertThat(success.get() + statusCounts.getOrDefault(409, 0)).isEqualTo(threads);
        assertThat(8 - success.get()).isEqualTo(stockAfter);
        assertThat(stockAfter >= 0).isTrue();
        System.out.println("U79 concurrentDistinct=PASS threads=" + threads + " ok=" + success.get()
                + " rejected=" + statusCounts.getOrDefault(409, 0) + " stockAfter=" + stockAfter);
    }

    @Test
    @Order(24)
    void midFailure_rollsBackEverything() {
        seedStock(LabFixtures.TENANT_T1, "SKU-RB", 10, "3.00");
        // 制造"扣减之后才失败"的真实中途失败：预置一条同键订单，使 insertOrder 撞 uk_order_idem。
        // 此时 deductStock 已在本事务内执行（10→8），随后唯一冲突 → 整个事务回滚。
        jdbc.update("INSERT INTO orders (tenant, sku, quantity, amount, status, idempotency_key)"
                + " VALUES ('t1','SKU-RB',1,3.00,'CREATED','rb-seed-key')");
        String body = """
                {"sku":"SKU-RB","quantity":2}
                """;
        ResponseEntity<String> response = post(body, token(LabFixtures.TENANT_T1), "rb-seed-key");
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.CONFLICT);
        assertThat(response.getBody()).contains("duplicate_request");
        // 不变量：扣减被回滚（库存回到10）、无幂等记录、只留下预置订单
        assertThat(jdbc.queryForObject(
                "SELECT available FROM stock WHERE tenant='t1' AND sku='SKU-RB'", Integer.class)).isEqualTo(10);
        assertThat(jdbc.queryForObject(
                "SELECT COUNT(*) FROM order_idempotency WHERE idempotency_key='rb-seed-key'", Integer.class))
                .isEqualTo(0);
        assertThat(jdbc.queryForObject(
                "SELECT COUNT(*) FROM orders WHERE idempotency_key='rb-seed-key'", Integer.class)).isEqualTo(1);
        System.out.println("U79 midFailureRollback=PASS deductedThenRolledBack stock=10 idemRows=0 orders=1");
    }

    @Test
    @Order(25)
    void exceptionTranslation_duplicateKeyPath() {
        // N02：DuplicateKey → DataIntegrityViolationException 分类已由 midFailure 验证；
        // 这里断言 MyBatis/Spring 转换确实产生 DuplicateKeyException 子类
        assertThat(org.springframework.dao.DataIntegrityViolationException.class)
                .isAssignableFrom(org.springframework.dao.DuplicateKeyException.class);
        System.out.println("U79 exceptionTranslation=PASS duplicateKeyIsIntegrityViolation=true");
    }

    @Test
    @Order(26)
    void lockTimeout_mapsTo503_notBusinessRejection() throws Exception {
        // N02/T03：占用行锁直到 InnoDB innodb_lock_wait_timeout（本会话默认50s太大），
        // 改为短超时会话：SET SESSION innodb_lock_wait_timeout=1，服务层 800ms 内不可得 → 503
        seedStock(LabFixtures.TENANT_T1, "SKU-LOCK", 10, "1.00");
        // 保持会话默认 innodb_lock_wait_timeout=50（不改全局/会话配置）。
        // 行锁被持有者占用；服务层 @Transactional(timeout=5) 使每条语句的 queryTimeout 先到（约5s），
        // Spring 翻译成 QueryTimeoutException → 503。服务器侧 innodb_lock_wait_timeout 分支未运行。
        try (Connection holder = DriverManager.getConnection(
                "jdbc:mysql://%s:%d/%s?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC"
                        .formatted(DB_HOST, DB_PORT, DB_NAME), DB_USER, DB_PASSWORD)) {
            holder.setAutoCommit(false);
            try (PreparedStatement lock = holder.prepareStatement(
                    "SELECT available FROM stock WHERE tenant='t1' AND sku='SKU-LOCK' FOR UPDATE")) {
                try (ResultSet rs = lock.executeQuery()) {
                    rs.next();
                }
                // 服务层事务在行锁上等待；结果应为 503 lock_wait_timeout，而不是 409
                long startedAt = System.nanoTime();
                ResponseEntity<String> response = post("""
                        {"sku":"SKU-LOCK","quantity":1}
                        """, token(LabFixtures.TENANT_T1), "lock-case");
                long elapsedMs = (System.nanoTime() - startedAt) / 1_000_000;
                assertThat(response.getStatusCode()).isEqualTo(HttpStatus.SERVICE_UNAVAILABLE);
                assertThat(response.getBody()).contains("lock_wait_timeout");
                assertThat(elapsedMs).isLessThan(10_000);
                System.out.println("U79 lockTimeout503=PASS viaQueryTimeout elapsedMs=" + elapsedMs);
            } finally {
                holder.rollback();
            }
        }
    }

    private static Map<?, ?> parse(String body) {
        try {
            var mapper = new tools.jackson.databind.json.JsonMapper();
            return mapper.readValue(body, Map.class);
        } catch (Exception ex) {
            throw new IllegalStateException(body, ex);
        }
    }
}