完善订单Webhook事件通知：按订单应用定向回调，新增签名与失败重试机制，并补充后台指派/转单通知触发

shaofan
1 parent 2449a3c0
Showing 8 changed files with 127 additions and 26 deletions
src/main/java/com/diligrp/rider/service/WebhookService.java
src/main/java/com/diligrp/rider/service/impl/AdminRiderServiceImpl.java
src/main/java/com/diligrp/rider/service/impl/DeliveryOrderServiceImpl.java
src/main/java/com/diligrp/rider/service/impl/DispatchServiceImpl.java
src/main/java/com/diligrp/rider/service/impl/RefundServiceImpl.java
src/main/java/com/diligrp/rider/service/impl/RiderOrderServiceImpl.java
src/main/java/com/diligrp/rider/service/impl/WebhookServiceImpl.java
src/main/java/com/diligrp/rider/task/OrderScheduleTask.java
 package com.diligrp.rider.service;
+import com.diligrp.rider.entity.Orders;
+
 /**
  * Webhook 推送服务
  * 订单状态变更时调用此服务通知接入方
@@ -13,6 +15,8 @@ public interface WebhookService {
      */
     void send(String event, Long bizId, String payload);
+    void sendOrderEvent(Orders order, String event, String payload);
+
     /** 重试失败的 Webhook */
     void retry(Long logId);
 }
@@ -17,8 +17,11 @@ import com.diligrp.rider.mapper.*;
 import com.diligrp.rider.service.AdminRiderService;
 import com.diligrp.rider.service.DeliveryOrderService;
 import com.diligrp.rider.service.RiderHoldLimitService;
+import com.diligrp.rider.service.WebhookService;
 import com.diligrp.rider.vo.DeliveryOrderCreateVO;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.DigestUtils;
@@ -31,6 +34,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+@Slf4j
 @Service
 @RequiredArgsConstructor
 public class AdminRiderServiceImpl implements AdminRiderService {
@@ -42,6 +46,8 @@ public class AdminRiderServiceImpl implements AdminRiderService {
     private final AdminScopeGuard adminScopeGuard;
     private final DeliveryOrderService deliveryOrderService;
     private final RiderHoldLimitService riderHoldLimitService;
+    private final WebhookService webhookService;
+    private final ObjectMapper objectMapper;
     @Override
     public void add(AdminRiderAddDTO dto, Long cityId) {
@@ -221,6 +227,7 @@ public class AdminRiderServiceImpl implements AdminRiderService {
         }
         int updated = ordersMapper.update(null, wrapper);
         if (updated == 0) throw new BizException("指派失败，请重试");
+        notifyOrderEvent(orderId, "order.dispatched");
     }
     @Override
@@ -248,6 +255,9 @@ public class AdminRiderServiceImpl implements AdminRiderService {
         }
         int updated = ordersMapper.update(null, wrapper);
         if (updated == 0) throw new BizException("操作失败，请重试");
+        if (trans == 1) {
+            notifyOrderEvent(orderId, "order.transferred");
+        }
     }
     @Override
@@ -266,6 +276,24 @@ public class AdminRiderServiceImpl implements AdminRiderService {
         deliveryOrderService.cancelByAdmin(order);
     }
+    private void notifyOrderEvent(Long orderId, String event) {
+        try {
+            Orders order = ordersMapper.selectById(orderId);
+            if (order == null || order.getAppKey() == null || order.getAppKey().isBlank()) return;
+            Map<String, Object> payload = new HashMap<>();
+            payload.put("event", event);
+            payload.put("outOrderNo", order.getOutOrderNo());
+            payload.put("deliveryOrderId", order.getId());
+            payload.put("orderNo", order.getOrderNo());
+            payload.put("status", order.getStatus());
+            payload.put("riderId", order.getRiderId());
+            payload.put("timestamp", System.currentTimeMillis() / 1000);
+            webhookService.sendOrderEvent(order, event, objectMapper.writeValueAsString(payload));
+        } catch (Exception e) {
+            log.warn("后台订单事件通知失败 orderId={} event={}", orderId, event, e);
+        }
+    }
+
     private String encryptPass(String pass) {
         return DigestUtils.md5DigestAsHex(pass.getBytes(StandardCharsets.UTF_8));
     }
@@ -303,7 +303,7 @@ public class DeliveryOrderServiceImpl implements DeliveryOrderService {
             payload.put("status", order.getStatus());
             payload.put("timestamp", System.currentTimeMillis() / 1000);
             String json = objectMapper.writeValueAsString(payload);
-            webhookService.send(event, order.getId(), json);
+            webhookService.sendOrderEvent(order, event, json);
         } catch (Exception e) {
             log.warn("通知回调失败 orderId={}", order.getId(), e);
         }
@@ -346,7 +346,7 @@ public class DispatchServiceImpl implements DispatchService {
             payload.put("riderId", order.getRiderId());
             payload.put("dispatchRiderId", order.getDispatchRiderId());
             payload.put("timestamp", System.currentTimeMillis() / 1000);
-            webhookService.send(event, orderId, objectMapper.writeValueAsString(payload));
+            webhookService.sendOrderEvent(order, event, objectMapper.writeValueAsString(payload));
         } catch (Exception e) {
             log.warn("派单事件通知失败 orderId={} event={}", orderId, event, e);
         }
@@ -70,6 +70,7 @@ public class RefundServiceImpl implements RefundService {
         ordersMapper.update(null, new LambdaUpdateWrapper<Orders>()
                 .eq(Orders::getId, orderId)
                 .set(Orders::getStatus, 7));
+        order.setStatus(7);
         // 写退款记录
         OrderRefundRecord record = new OrderRefundRecord();
@@ -115,12 +116,14 @@ public class RefundServiceImpl implements RefundService {
             ordersMapper.update(null, new LambdaUpdateWrapper<Orders>()
                     .eq(Orders::getId, record.getOid())
                     .set(Orders::getStatus, 8));
+            order.setStatus(8);
             notifyRefundEvent(order, "order.refund_success");
         } else {
             // 退款拒绝 → 订单状态改为退款拒绝
             ordersMapper.update(null, new LambdaUpdateWrapper<Orders>()
                     .eq(Orders::getId, record.getOid())
                     .set(Orders::getStatus, 9));
+            order.setStatus(9);
             notifyRefundEvent(order, "order.refund_reject");
         }
     }
@@ -148,7 +151,7 @@ public class RefundServiceImpl implements RefundService {
             payload.put("orderNo", order.getOrderNo());
             payload.put("status", order.getStatus());
             payload.put("timestamp", System.currentTimeMillis() / 1000);
-            webhookService.send(event, order.getId(), objectMapper.writeValueAsString(payload));
+            webhookService.sendOrderEvent(order, event, objectMapper.writeValueAsString(payload));
         } catch (Exception e) {
             log.warn("退款事件通知失败 orderId={}", order.getId(), e);
         }
@@ -256,7 +256,7 @@ public class RiderOrderServiceImpl implements RiderOrderService {
             payload.put("status", order.getStatus());
             payload.put("riderId", order.getRiderId());
             payload.put("timestamp", System.currentTimeMillis() / 1000);
-            webhookService.send(event, orderId, objectMapper.writeValueAsString(payload));
+            webhookService.sendOrderEvent(order, event, objectMapper.writeValueAsString(payload));
         } catch (Exception e) {
             log.warn("订单事件通知失败 orderId={} event={}", orderId, event, e);
         }
@@ -414,7 +414,7 @@ public class RiderOrderServiceImpl implements RiderOrderService {
         if (order.getStatus() == 3) {
             // status=3（已接单，未取货）：直接回抢单池，无需审批
-            ordersMapper.update(null, new LambdaUpdateWrapper<Orders>()
+            int updated = ordersMapper.update(null, new LambdaUpdateWrapper<Orders>()
                     .eq(Orders::getId, orderId)
                     .eq(Orders::getRiderId, riderId)
                     .eq(Orders::getStatus, 3)
@@ -426,6 +426,8 @@ public class RiderOrderServiceImpl implements RiderOrderService {
                     .set(Orders::getRiderIncome, BigDecimal.ZERO)
                     .set(Orders::getSubstationIncome, BigDecimal.ZERO)
                     .set(Orders::getTransTime, now));
+            if (updated == 0) throw new BizException(980, "操作失败");
+            notifyOrderEvent(orderId, "order.transferred");
         } else {
             // status=4（取货后配送中）：需分站审批，设为申请中
             ordersMapper.update(null, new LambdaUpdateWrapper<Orders>()
@@ -3,13 +3,17 @@ package com.diligrp.rider.service.impl;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.diligrp.rider.entity.OpenApp;
+import com.diligrp.rider.entity.Orders;
 import com.diligrp.rider.entity.WebhookLog;
 import com.diligrp.rider.mapper.OpenAppMapper;
+import com.diligrp.rider.mapper.OrdersMapper;
 import com.diligrp.rider.mapper.WebhookLogMapper;
 import com.diligrp.rider.service.WebhookService;
+import com.diligrp.rider.util.SignUtil;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import java.net.URI;
@@ -18,6 +22,7 @@ import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.List;
+import java.util.UUID;
 @Slf4j
 @Service
@@ -25,6 +30,7 @@ import java.util.List;
 public class WebhookServiceImpl implements WebhookService {
     private final OpenAppMapper openAppMapper;
+    private final OrdersMapper ordersMapper;
     private final WebhookLogMapper webhookLogMapper;
     private final ObjectMapper objectMapper;
@@ -35,47 +41,102 @@ public class WebhookServiceImpl implements WebhookService {
     @Override
     @Async
     public void send(String event, Long bizId, String payload) {
-        // 查找订阅该事件的所有应用
-        List<OpenApp> apps = openAppMapper.selectList(
-                new LambdaQueryWrapper<OpenApp>().eq(OpenApp::getStatus, 1));
-
-        for (OpenApp app : apps) {
-            if (app.getWebhookUrl() == null || app.getWebhookUrl().isBlank()) continue;
-            if (!isSubscribed(app.getWebhookEvents(), event)) continue;
-            doSend(app, event, bizId, payload, 0);
+        Orders order = ordersMapper.selectById(bizId);
+        if (order == null) {
+            log.warn("Webhook 发送跳过，订单不存在 bizId={} event={}", bizId, event);
+            return;
         }
+        sendOrderEvent(order, event, payload);
+    }
+
+    @Override
+    @Async
+    public void sendOrderEvent(Orders order, String event, String payload) {
+        if (order == null || order.getAppKey() == null || order.getAppKey().isBlank()) return;
+        OpenApp app = openAppMapper.selectOne(new LambdaQueryWrapper<OpenApp>()
+                .eq(OpenApp::getAppKey, order.getAppKey())
+                .eq(OpenApp::getStatus, 1)
+                .last("LIMIT 1"));
+        if (app == null) return;
+        if (!isSubscribed(app.getWebhookEvents(), event)) return;
+
+        String url = order.getCallbackUrl();
+        if (url == null || url.isBlank()) {
+            url = app.getWebhookUrl();
+        }
+        if (url == null || url.isBlank()) return;
+
+        doSend(app, url, event, order.getId(), payload, 0);
     }
     @Override
     public void retry(Long logId) {
         WebhookLog log = webhookLogMapper.selectById(logId);
         if (log == null || log.getStatus() == 1) return;
-        OpenApp app = openAppMapper.selectById(log.getAppId());
+        retry(log);
+    }
+
+    @Scheduled(fixedDelay = 300_000)
+    public void retryFailedWebhooks() {
+        List<WebhookLog> logs = webhookLogMapper.selectList(new LambdaQueryWrapper<WebhookLog>()
+                .eq(WebhookLog::getStatus, 0)
+                .lt(WebhookLog::getRetryCount, 5)
+                .orderByAsc(WebhookLog::getCreateTime)
+                .last("LIMIT 20"));
+        for (WebhookLog log : logs) {
+            retry(log);
+        }
+    }
+
+    private void retry(WebhookLog webhookLog) {
+        OpenApp app = openAppMapper.selectById(webhookLog.getAppId());
         if (app == null) return;
-        doSend(app, log.getEvent(), log.getBizId(), log.getPayload(), log.getRetryCount() + 1);
+        String url = webhookLog.getUrl();
+        if (url == null || url.isBlank()) {
+            url = app.getWebhookUrl();
+        }
+        if (url == null || url.isBlank()) return;
+        SendResult result = sendHttp(app, url, webhookLog.getEvent(), webhookLog.getPayload());
+        webhookLog.setUrl(url);
+        webhookLog.setResponseCode(result.responseCode());
+        webhookLog.setResponseBody(trimResponseBody(result.responseBody()));
+        webhookLog.setStatus(result.status());
+        webhookLog.setRetryCount((webhookLog.getRetryCount() == null ? 0 : webhookLog.getRetryCount()) + 1);
+        webhookLogMapper.updateById(webhookLog);
     }
-    private void doSend(OpenApp app, String event, Long bizId, String payload, int retryCount) {
+    private void doSend(OpenApp app, String url, String event, Long bizId, String payload, int retryCount) {
+        SendResult result = sendHttp(app, url, event, payload);
         WebhookLog webhookLog = new WebhookLog();
         webhookLog.setAppId(app.getId());
         webhookLog.setEvent(event);
         webhookLog.setBizId(bizId);
-        webhookLog.setUrl(app.getWebhookUrl());
+        webhookLog.setUrl(url);
         webhookLog.setPayload(payload);
         webhookLog.setRetryCount(retryCount);
         webhookLog.setCreateTime(System.currentTimeMillis() / 1000);
+        webhookLog.setResponseCode(result.responseCode());
+        webhookLog.setResponseBody(trimResponseBody(result.responseBody()));
+        webhookLog.setStatus(result.status());
+        webhookLogMapper.insert(webhookLog);
+    }
+    private SendResult sendHttp(OpenApp app, String url, String event, String payload) {
         int responseCode = 0;
         String responseBody = "";
         int status = 0;
         try {
+            String timestamp = String.valueOf(System.currentTimeMillis() / 1000);
+            String nonce = UUID.randomUUID().toString().replace("-", "");
             HttpRequest request = HttpRequest.newBuilder()
-                    .uri(URI.create(app.getWebhookUrl()))
+                    .uri(URI.create(url))
                     .header("Content-Type", "application/json")
                     .header("X-App-Key", app.getAppKey())
                     .header("X-Event", event)
-                    .header("X-Timestamp", String.valueOf(System.currentTimeMillis() / 1000))
+                    .header("X-Timestamp", timestamp)
+                    .header("X-Nonce", nonce)
+                    .header("X-Sign", SignUtil.sign(app.getAppKey(), timestamp, nonce, app.getAppSecret()))
                     .POST(HttpRequest.BodyPublishers.ofString(payload))
                     .timeout(Duration.ofSeconds(10))
                     .build();
@@ -83,19 +144,22 @@ public class WebhookServiceImpl implements WebhookService {
             HttpResponse<String> response = HTTP_CLIENT.send(request,
                     HttpResponse.BodyHandlers.ofString());
             responseCode = response.statusCode();
-            responseBody = response.body();
+            responseBody = response.body() == null ? "" : response.body();
             if (responseCode == 200) status = 1;
         } catch (Exception e) {
             log.warn("Webhook 发送失败 appId={} event={} err={}", app.getId(), event, e.getMessage());
-            responseBody = e.getMessage();
+            responseBody = e.getMessage() == null ? "" : e.getMessage();
         }
+        return new SendResult(responseCode, responseBody, status);
+    }
-        webhookLog.setResponseCode(responseCode);
-        webhookLog.setResponseBody(responseBody.length() > 500 ? responseBody.substring(0, 500) : responseBody);
-        webhookLog.setStatus(status);
-        webhookLogMapper.insert(webhookLog);
+    private String trimResponseBody(String responseBody) {
+        if (responseBody == null) return "";
+        return responseBody.length() > 500 ? responseBody.substring(0, 500) : responseBody;
     }
+    private record SendResult(int responseCode, String responseBody, int status) {}
+
     /** 检查应用是否订阅了某事件 */
     private boolean isSubscribed(String webhookEvents, String event) {
         if (webhookEvents == null || webhookEvents.isBlank()) return false;
@@ -123,7 +123,7 @@ public class OrderScheduleTask {
             payload.put("status", 10);
             payload.put("reason", "超时无人接单，系统自动取消");
             payload.put("timestamp", System.currentTimeMillis() / 1000);
-            webhookService.send("order.cancelled", order.getId(),
+            webhookService.sendOrderEvent(order, "order.cancelled",
                     objectMapper.writeValueAsString(payload));
         } catch (Exception e) {
             log.warn("取消通知失败 orderId={}", order.getId(), e);