flow-center-service/src/test/java/com/yunhai/flowcenter/callback/CallbackDispatchIT.java

package com.yunhai.flowcenter.callback;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import com.yunhai.flowcenter.entity.FlwCallbackLog;
import com.yunhai.flowcenter.model.OapiModels;
import com.yunhai.flowcenter.entity.FlwCallbackReg;
import com.yunhai.flowcenter.enums.FlwEnums;
import com.yunhai.flowcenter.repository.FlwCallbackRegRepository;
import com.yunhai.flowcenter.service.FlowInstanceService;
import com.yunhai.flowcenter.service.OapiFlowService;
import com.yunhai.flowcenter.support.FlowTestSupport;
import com.yunhai.flowcenter.support.OapiSignedClient;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.web.server.LocalServerPort;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.test.context.ActiveProfiles;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * 回调分发 IT：注册（探活）→ 代发起 → AFTER_COMMIT 事件 → 落 flw_callback_log →
 * 异步推送 X-Flw-* 四头 → stub 接收端验签 + payload 字段断言；推送失败持续重试（T1.1：
 * 越过退避表封顶 1h 不死信、挂起告警恰一次、瞬时失败自动治愈）；挂起可视/标记死信/rePush 复活（T1.2）。
 * 注意：本类必须单类独跑（平台 IT 判据）。
 */
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@ActiveProfiles("test")
class CallbackDispatchIT extends FlowTestSupport {

    private static final String APP_KEY = "100018";
    private static final String APP_SECRET = "svc-hall-dev-secret-100018";

    @LocalServerPort
    private int port;

    @Autowired
    private OapiFlowService oapiFlowService;
    @Autowired
    private FlowInstanceService instanceService;
    @Autowired
    private CallbackPushService pushService;
    @Autowired
    private com.yunhai.flowcenter.service.FlowCallbackAdminService callbackAdminService;
    @Autowired
    private FlwCallbackRegRepository regRepository;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private com.yunhai.platform.appauth.LocalAppIdentityProvider appIdentity;

    private HttpServer stubServer;
    private int stubPort;
    private final CopyOnWriteArrayList<Received> received = new CopyOnWriteArrayList<>();

    /** 计数版 UserMsgClient：只数挂起告警（hang:* msgNo），不真正外发（@Primary 注入 CallbackPushService） */
    static final CountingUserMsgClient STUB_USER_MSG = new CountingUserMsgClient();

    @TestConfiguration
    static class StubConfig {
        @Bean
        @Primary
        public com.yunhai.flowcenter.stub.StubUserClient stubUserClient() {
            return new com.yunhai.flowcenter.stub.StubUserClient();
        }

        @Bean
        @Primary
        public CountingUserMsgClient countingUserMsgClient() {
            return STUB_USER_MSG;
        }
    }

    static class CountingUserMsgClient extends com.yunhai.flowcenter.notify.UserMsgClient {
        final java.util.concurrent.atomic.AtomicInteger hangAlerts =
                new java.util.concurrent.atomic.AtomicInteger();

        CountingUserMsgClient() {
            // enabled=false：本用例只覆盖回调派发，站内信为哑实现（两个方法均覆写）
            super(null, null, false, "");
        }

        @Override
        public void sendUserMsg(Integer schId, List<Integer> userIds, Integer msgType, String bizType,
                                String bizId, String title, String content, String jumpPath, String msgNo,
                                Integer expireDays) {
            if (msgNo != null && msgNo.startsWith("hang:")) {
                hangAlerts.incrementAndGet();
            }
        }
    }

    static class Received {
        String body;
        String eventId;
        String timestamp;
        String nonce;
        String signature;
        /** 应用签名三件套（供 /oapi/* 强制验签的接收方，与 X-Flw-* 并存） */
        String appKey;
        String appTimestamp;
        String appSignature;
    }

    @BeforeEach
    void setUp() throws Exception {
        cleanDb();
        received.clear();
        STUB_USER_MSG.hangAlerts.set(0);
        stubServer = HttpServer.create(new InetSocketAddress(0), 0);
        stubPort = stubServer.getAddress().getPort();
        stubServer.createContext("/callback/flow", exchange -> {
            Received r = new Received();
            r.body = readBody(exchange);
            r.eventId = exchange.getRequestHeaders().getFirst("X-Flw-EventId");
            r.timestamp = exchange.getRequestHeaders().getFirst("X-Flw-Timestamp");
            r.nonce = exchange.getRequestHeaders().getFirst("X-Flw-Nonce");
            r.signature = exchange.getRequestHeaders().getFirst("X-Flw-Signature");
            r.appKey = exchange.getRequestHeaders().getFirst("X-App-Key");
            r.appTimestamp = exchange.getRequestHeaders().getFirst("X-Timestamp");
            r.appSignature = exchange.getRequestHeaders().getFirst("X-Signature");
            received.add(r);
            byte[] resp = "{\"ok\":true}".getBytes(StandardCharsets.UTF_8);
            exchange.sendResponseHeaders(200, resp.length);
            exchange.getResponseBody().write(resp);
            exchange.close();
        });
        stubServer.start();
    }

    @AfterEach
    void tearDown() {
        stubServer.stop(0);
    }

    @Test
    void fullChain_register_apply_pushesSignedEvents() {
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t20", "回调全链", linearDef("ybb", "审核"));
        // 1. 注册（探活命中 stub）→ secret 下发
        OapiModels.CallbackRegResultVo reg = oapiFlowService.registerCallback(SCH, APP_KEY,
                regBody(def.getFlowGuid(), "http://localhost:" + stubPort + "/callback/flow", null));
        assertTrue(reg.getSecret() != null && reg.getSecret().length() >= 16);

        // 2. 代发起 → flow.submitted + flow.task.created 两事件先后落库并推送
        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-1\",\"bizTitle\":\"张三的回调件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        org.springframework.http.ResponseEntity<JsonNode> applyResp =
                client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);
        // 3. 轮询 stub：submitted + task.created 两事件都到齐（探活 ping 不计）
        waitUntil(() -> received.stream().anyMatch(r -> r.body.contains("flow.submitted"))
                && received.stream().anyMatch(r -> r.body.contains("flow.task.created")), 10_000);
        boolean hasSubmitted = false;
        boolean hasTaskCreated = false;
        for (Received r : received) {
            JsonNode payload = readJson(r.body);
            String event = payload.get("event").asText();
            if ("ping".equals(event)) {
                continue; // 注册探活 ping 不参与事件协议断言
            }
            if ("flow.submitted".equals(event)) {
                hasSubmitted = true;
            }
            if ("flow.task.created".equals(event)) {
                hasTaskCreated = true;
            }
            // 协议断言：必带字段
            assertEquals("1", payload.get("schId").asText());
            assertEquals(def.getFlowGuid(), payload.get("flowGuid").asText());
            assertEquals("svc_item", payload.get("bizType").asText());
            assertEquals("cb-1", payload.get("bizId").asText());
            assertEquals("张三的回调件", payload.get("bizTitle").asText());
            assertTrue(payload.get("instanceVersion").asInt() >= 1, "instanceVersion 应为办件乐观锁版本");
            assertNotNull(payload.get("eventId"));
            assertNotNull(payload.get("occurredTime"));
            // 2026-09-20 信封升级：taskId + 操作人（发起直达链路 → 操作人=发起人快照）
            assertEquals("201", payload.get("operatorUserId").asText(), "operatorUserId 应为发起人");
            assertEquals("张三", payload.get("operatorName").asText(), "operatorName 应为发起人");
            if ("flow.task.created".equals(event)) {
                assertTrue(payload.get("taskId") != null && payload.get("taskId").isNumber(),
                        "task.created 必带顶层 taskId");
            } else if ("flow.submitted".equals(event)) {
                assertTrue(payload.get("taskId") == null || payload.get("taskId").isNull(),
                        "办件级事件 taskId 应为 null");
            }
            // 签名断言：Base64(HmacSHA256(secret, ts.nonce.body))
            String expected = hmac(reg.getSecret(), r.timestamp, r.nonce, r.body);
            assertEquals(expected, r.signature, "推送必须带合法 X-Flw-Signature");
            assertEquals(r.eventId, payload.get("eventId").asText(), "头与体 eventId 一致");
            // 应用签名断言：接收方把端点收在 /oapi/* 下时由平台强制验签（未申报 basePath → 签整个 path）
            assertAppSignature("/callback/flow", r);
        }
        assertTrue(hasSubmitted, "应有 flow.submitted 事件");
        assertTrue(hasTaskCreated, "应有 flow.task.created 事件");
        // 注册探活 ping 也必须带应用签名（否则接收方 401，RestTemplate 对 4xx 抛异常会误判不可达）
        Received ping = received.stream()
                .filter(r -> r.body != null && r.body.contains("\"ping\"")).findFirst().orElse(null);
        assertNotNull(ping, "注册探活应命中 stub");
        assertAppSignature("/callback/flow", ping);
        // 4. 日志表成功态（异步推送落定，轮询等待）
        assertTrue(logRepository.findAll().size() >= 2);
        waitUntil(() -> logRepository.findAll().stream()
                .allMatch(l -> l.getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode()), 10_000);
    }

    /**
     * 应用签名 canonical：注册申报 basePath（接收方 context-path）时，先从 callback_url 的 path 剥离
     * 再签名——与接收方 AppAuthFilter 用 request.contextPath 剥离后的 path 对齐。
     * 生产 Notice 部署在 /notice 前缀下（deploy/notice.yaml context-path）即此场景。
     */
    @Test
    void appSignature_canonicalTargetStripsRegisteredBasePath() {
        CopyOnWriteArrayList<Received> hits = new CopyOnWriteArrayList<>();
        stubServer.createContext("/notice/oapi/notice/flow/callback", exchange -> {
            Received r = new Received();
            r.body = readBody(exchange);
            r.appKey = exchange.getRequestHeaders().getFirst("X-App-Key");
            r.appTimestamp = exchange.getRequestHeaders().getFirst("X-Timestamp");
            r.appSignature = exchange.getRequestHeaders().getFirst("X-Signature");
            hits.add(r);
            byte[] resp = "{\"ok\":true}".getBytes(StandardCharsets.UTF_8);
            exchange.sendResponseHeaders(200, resp.length);
            exchange.getResponseBody().write(resp);
            exchange.close();
        });
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t25", "基路径签名", linearDef("ybb", "审核"));
        OapiModels.CallbackRegBody rb = regBody(def.getFlowGuid(),
                "http://localhost:" + stubPort + "/notice/oapi/notice/flow/callback", "flow.submitted");
        rb.setBasePath("/notice");
        oapiFlowService.registerCallback(SCH, APP_KEY, rb);

        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-5\",\"bizTitle\":\"基路径件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);

        waitUntil(() -> hits.stream().anyMatch(r -> r.body != null && r.body.contains("flow.submitted")), 10_000);
        Received push = hits.stream()
                .filter(r -> r.body != null && r.body.contains("flow.submitted"))
                .findFirst().orElseThrow(IllegalStateException::new);
        assertAppSignature("/oapi/notice/flow/callback", push);
    }

    @Test
    void pushFailure_persistsRetryBeyondBackoffList() {
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t21", "持续重试", linearDef("ybb", "审核"));
        // 只订阅 flow.submitted：单事件单日志行，断言确定
        oapiFlowService.registerCallback(SCH, APP_KEY,
                regBody(def.getFlowGuid(), "http://localhost:" + stubPort + "/callback/flow", "flow.submitted"));
        FlwCallbackReg regRow = regRepository.findBySchIdAndAppKeyAndFlowGuid(SCH, APP_KEY, def.getFlowGuid())
                .orElseThrow(IllegalStateException::new);
        // 注册指向死端口模拟业务方宕机（探活已过，直接改库）
        regRow.setCallbackUrl("http://localhost:1/dead");
        regRepository.save(regRow);

        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-2\",\"bizTitle\":\"张三的挂起件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);

        // 首推失败：retryCount=1（异步首推，轮询等待）
        waitUntil(() -> logRepository.findAll().stream()
                .anyMatch(l -> l.getRetryCount() != null && l.getRetryCount() >= 1), 10_000);
        // 手工逐次触发扫描语义再失败 5 轮（共 6 次，越过退避表 5 项）
        for (int round = 0; round < 5; round++) {
            for (FlwCallbackLog row : logRepository.findAll()) {
                pushService.pushAsync(row.getId());
            }
        }
        FlwCallbackLog hung = logRepository.findAll().get(0);
        // T1.1 核心语义：持续重试不自动死信——status 仍=0、retryCount=6、nextRetry≈now+60min（封顶）
        assertEquals(FlwEnums.CallbackLogStatus.PENDING.getCode(), hung.getStatus(),
                "越过退避表后必须保持待推持续重试，不得自动死信");
        assertEquals(6, hung.getRetryCount());
        assertNotNull(hung.getNextRetryTime());
        java.time.LocalDateTime expectedNext = java.time.LocalDateTime.now().plusMinutes(60);
        assertTrue(hung.getNextRetryTime().isAfter(expectedNext.minusMinutes(5))
                        && hung.getNextRetryTime().isBefore(expectedNext.plusMinutes(5)),
                "越过退避表后 nextRetryTime 应封顶 +60 分钟: " + hung.getNextRetryTime());
        assertTrue(logRepository.findAll().stream()
                .allMatch(l -> l.getStatus() == FlwEnums.CallbackLogStatus.PENDING.getCode()),
                "任何行都不应出现死信（status=2）");
        assertNotNull(hung.getLastError());
    }

    @Test
    void hangAlertSentExactlyOnce() {
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t23", "挂起告警", linearDef("ybb", "审核"));
        oapiFlowService.registerCallback(SCH, APP_KEY,
                regBody(def.getFlowGuid(), "http://localhost:" + stubPort + "/callback/flow", "flow.submitted"));
        FlwCallbackReg regRow = regRepository.findBySchIdAndAppKeyAndFlowGuid(SCH, APP_KEY, def.getFlowGuid())
                .orElseThrow(IllegalStateException::new);
        regRow.setCallbackUrl("http://localhost:1/dead");
        regRepository.save(regRow);

        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-4\",\"bizTitle\":\"张三的告警件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);
        waitUntil(() -> logRepository.findAll().stream()
                .anyMatch(l -> l.getRetryCount() != null && l.getRetryCount() >= 1), 10_000);

        // 失败到第 6 次（=挂起阈值）：告警恰一次
        for (int round = 0; round < 5; round++) {
            for (FlwCallbackLog row : logRepository.findAll()) {
                pushService.pushAsync(row.getId());
            }
        }
        assertEquals(1, STUB_USER_MSG.hangAlerts.get(), "第 6 次失败后挂起告警应恰发送一次");

        // 第 7 次失败：仍是 1（恰一次）
        for (FlwCallbackLog row : logRepository.findAll()) {
            pushService.pushAsync(row.getId());
        }
        assertEquals(1, STUB_USER_MSG.hangAlerts.get(), "第 7 次失败后告警计数仍应为 1");
        assertEquals(7, logRepository.findAll().get(0).getRetryCount());
    }

    @Test
    void transientFailureRecovers() {
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t24", "瞬时失败治愈", linearDef("ybb", "审核"));
        // 抖动端点：探活(第1次) 200、首推(第2次) 500、其后 200——模拟消费方瞬时故障后恢复
        AtomicInteger flakyCount = new AtomicInteger(0);
        stubServer.createContext("/callback/flaky", exchange -> {
            boolean fail = flakyCount.incrementAndGet() == 2;
            byte[] resp = "{\"ok\":true}".getBytes(StandardCharsets.UTF_8);
            exchange.sendResponseHeaders(fail ? 500 : 200, resp.length);
            exchange.getResponseBody().write(resp);
            exchange.close();
        });
        oapiFlowService.registerCallback(SCH, APP_KEY,
                regBody(def.getFlowGuid(), "http://localhost:" + stubPort + "/callback/flaky", "flow.submitted"));

        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-5\",\"bizTitle\":\"张三的抖动件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);

        // 首推 500 → 进入重试（status=0, retryCount=1）
        waitUntil(() -> logRepository.findAll().stream()
                .anyMatch(l -> l.getRetryCount() != null && l.getRetryCount() >= 1), 10_000);
        // 消费方恢复后（扫描/重推触发）→ 200 → status=1 + pushTime 落定
        for (FlwCallbackLog row : logRepository.findAll()) {
            pushService.pushAsync(row.getId());
        }
        FlwCallbackLog recovered = logRepository.findAll().get(0);
        assertEquals(FlwEnums.CallbackLogStatus.SUCCESS.getCode(), recovered.getStatus(),
                "瞬时失败恢复后应自动治愈为成功");
        assertNotNull(recovered.getPushTime());
    }

    @Test
    void hungList_paged_andMarkDead_thenRepushRevives() {
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t25", "挂起管理", linearDef("ybb", "审核"));
        oapiFlowService.registerCallback(SCH, APP_KEY,
                regBody(def.getFlowGuid(), "http://localhost:" + stubPort + "/callback/flow", "flow.submitted"));
        FlwCallbackReg regRow = regRepository.findBySchIdAndAppKeyAndFlowGuid(SCH, APP_KEY, def.getFlowGuid())
                .orElseThrow(IllegalStateException::new);
        regRow.setCallbackUrl("http://localhost:1/dead");
        regRepository.save(regRow);

        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-6\",\"bizTitle\":\"张三的管理件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);
        waitUntil(() -> logRepository.findAll().stream()
                .anyMatch(l -> l.getRetryCount() != null && l.getRetryCount() >= 1), 10_000);
        for (int round = 0; round < 5; round++) {
            for (FlwCallbackLog row : logRepository.findAll()) {
                pushService.pushAsync(row.getId());
            }
        }
        FlwCallbackLog hung = logRepository.findAll().get(0);
        assertEquals(6, hung.getRetryCount());

        // 挂起列表：status=0 且 retryCount≥阈值的分页行，不含正常行
        com.yunhai.platform.utils.model.BasePageResult<FlwCallbackLog> hungPage =
                callbackAdminService.pageHung(SCH, 1, 10);
        assertTrue(hungPage.getContent().stream().anyMatch(l -> l.getId().equals(hung.getId())));
        assertTrue(hungPage.getContent().stream()
                .allMatch(l -> l.getStatus() == FlwEnums.CallbackLogStatus.PENDING.getCode()
                        && l.getRetryCount() >= 6));

        // 管理端手工标记死信（唯一死信入口）
        callbackAdminService.markDead(SCH, hung.getId());
        FlwCallbackLog dead = logRepository.findById(hung.getId()).orElseThrow(IllegalStateException::new);
        assertEquals(FlwEnums.CallbackLogStatus.DEAD.getCode(), dead.getStatus());

        // 恢复消费方地址后既有 rePush 复活：置回待推立即推送 → 成功
        regRow.setCallbackUrl("http://localhost:" + stubPort + "/callback/flow");
        regRepository.save(regRow);
        callbackAdminService.rePush(SCH, java.util.Collections.singletonList(hung.getId()));
        waitUntil(() -> {
            FlwCallbackLog row = logRepository.findById(hung.getId()).orElse(null);
            return row != null && row.getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode();
        }, 10_000);
    }

    @Test
    void unsubscribedEvent_notDelivered() throws InterruptedException {
        mapDept("ybb", 101, "甲");
        com.yunhai.flowcenter.entity.FlwDef def = saveAndPublishDef("t22", "订阅过滤", linearDef("ybb", "审核"));
        // 只订阅 flow.completed：submitted/task.created 不推送
        oapiFlowService.registerCallback(SCH, APP_KEY,
                regBody(def.getFlowGuid(), "http://localhost:" + stubPort + "/callback/flow",
                        "flow.completed"));
        OapiSignedClient client = new OapiSignedClient("http://localhost:" + port, APP_KEY, APP_SECRET);
        String body = "{"
                + "\"flowGuid\":\"" + def.getFlowGuid() + "\","
                + "\"bizType\":\"svc_item\",\"bizId\":\"cb-3\",\"bizTitle\":\"张三的订阅件\","
                + "\"submitterId\":201,\"submitterName\":\"张三\"}";
        client.post("/oapi/flow/apply?schId=1", body, JsonNode.class);
        Thread.sleep(1500);
        // stub 收到的注册探活 ping 不算事件推送
        long eventPushes = received.stream()
                .filter(r -> !r.body.contains("\"event\":\"ping\"")).count();
        assertEquals(0, eventPushes, "未订阅事件不应推送");
    }

    /** 组注册请求体（registerCallback 返回下发 secret） */
    private OapiModels.CallbackRegBody regBody(String flowGuid, String url, String eventTypes) {
        OapiModels.CallbackRegBody body = new OapiModels.CallbackRegBody();
        body.setFlowGuid(flowGuid);
        body.setCallbackUrl(url);
        body.setEventTypes(eventTypes);
        return body;
    }

    /** 应用签名断言：canonical = METHOD + "\n" + 剥离 base-path 后的 path(+排序 query) + "\n" + ts */
    private void assertAppSignature(String expectedTarget, Received r) {
        assertNotNull(r.appKey, "推送必须带 X-App-Key");
        assertEquals(appIdentity.getAppKey(), r.appKey, "签名身份应是本应用 appKey");
        assertNotNull(r.appTimestamp, "推送必须带 X-Timestamp");
        assertNotNull(r.appSignature, "推送必须带 X-Signature");
        assertTrue(com.yunhai.platform.utils.crypto.SignUtil.verifyRequest(
                        "POST", expectedTarget, r.appTimestamp, appIdentity.getAppSecret(), r.appSignature),
                "应用签名应按 target=" + expectedTarget + " 可验（与接收方 AppAuthFilter 剥离 context-path 后对齐）");
    }

    /** 轮询等待（替代 Awaitility，离线环境无该依赖） */
    private static void waitUntil(java.util.function.BooleanSupplier cond, long timeoutMs) {
        long deadline = System.currentTimeMillis() + timeoutMs;
        while (System.currentTimeMillis() < deadline) {
            if (cond.getAsBoolean()) {
                return;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }
        org.junit.jupiter.api.Assertions.assertTrue(cond.getAsBoolean(), "等待条件超时");
    }

    private static String readBody(HttpExchange exchange) {
        try (InputStream in = exchange.getRequestBody()) {
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            byte[] buf = new byte[4096];
            int n;
            while ((n = in.read(buf)) > 0) {
                out.write(buf, 0, n);
            }
            return out.toString("UTF-8");
        } catch (Exception e) {
            return "";
        }
    }

    private JsonNode readJson(String s) {
        try {
            return objectMapper.readTree(s);
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }

    private static String hmac(String secret, String ts, String nonce, String body) {
        try {
            Mac mac = Mac.getInstance("HmacSHA256");
            mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
            byte[] raw = mac.doFinal((ts + "." + nonce + "." + body).getBytes(StandardCharsets.UTF_8));
            return Base64.getEncoder().encodeToString(raw);
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }
}