充值订单关闭，余额支付退款，接收支付结果广播，推送mqtt

FullCardClient/src/main/java/com/zhongshu/card/client/model/pay/RechargeOrderModel.java

@@ -29,4 +29,7 @@ public class RechargeOrderModel {
 
     @Schema(description = "订单号")
     private String orderNo;
+
+    @Schema(description = "描述")
+    private String description;
 }

FullCardServer/src/main/java/com/zhongshu/card/server/core/controller/pay/balance/RechargeController.java

@@ -1,5 +1,6 @@
 package com.zhongshu.card.server.core.controller.pay.balance;
 
+import com.github.microservice.pay.client.ret.ResultContent;
 import com.zhongshu.card.client.model.pay.QueryRechargeParam;
 import com.zhongshu.card.client.model.pay.RechargeParam;
 import com.zhongshu.card.server.core.service.pay.BalancePayService;
@@ -30,7 +31,19 @@ public class RechargeController {
 
     @Operation(summary = "查询用户余额", description = "查询用户余额")
     @RequestMapping(value = "queryUserBalance", method = RequestMethod.GET)
-    public Object queryUserBalance(@RequestParam("projectOid") String projectOid, @RequestParam("userId") String userId){
+    public Object queryUserBalance(@RequestParam(value = "projectOid", required = false) String projectOid, @RequestParam(value = "userId", required = false) String userId){
         return balancePayService.queryUserBalance(projectOid, userId);
     }
+
+    @Operation(summary = "查询充值订单详情", description = "查询充值订单详情")
+    @RequestMapping(value = "queryInfo", method = RequestMethod.GET)
+    public Object queryInfo(@RequestParam("orderNo") String orderNo){
+        return balancePayService.queryInfo(orderNo);
+    }
+
+    @Operation(summary = "关闭充值订单", description = "关闭充值订单（只有待支付状态时可以关闭）")
+    @RequestMapping(value = "close", method = RequestMethod.GET)
+    public Object close(@RequestParam("orderNo") String orderNo){
+        return balancePayService.close(orderNo);
+    }
 }

FullCardServer/src/main/java/com/zhongshu/card/server/core/dao/pay/BalanceRechargeOrderDao.java

@@ -7,4 +7,6 @@ import com.zhongshu.card.server.core.domain.pay.BalanceRechargeOrder;
 public interface BalanceRechargeOrderDao extends MongoDao<BalanceRechargeOrder>, BalanceRechargeOrderDaoExtend {
 
     BalanceRechargeOrder findTopByOrderNo(String orderNo);
+
+    BalanceRechargeOrder findTopById(String id);
 }

FullCardServer/src/main/java/com/zhongshu/card/server/core/dao/pay/extend/BalanceRechargeOrderDaoExtend.java

@@ -1,11 +1,23 @@
 package com.zhongshu.card.server.core.dao.pay.extend;
 
+import ch.qos.logback.core.util.StringUtil;
+import com.github.microservice.types.OrderState;
 import com.zhongshu.card.client.model.pay.QueryRechargeParam;
+import com.zhongshu.card.client.type.payAccount.RechargeOrderStatus;
 import com.zhongshu.card.server.core.domain.pay.BalanceRechargeOrder;
+import com.zhongshu.payment.client.model.TotalStatisticsModel;
+import com.zhongshu.payment.client.types.TradeType;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.query.Criteria;
 
+import java.util.ArrayList;
 import java.util.List;
 
 public interface BalanceRechargeOrderDaoExtend {
 
     List<BalanceRechargeOrder> listByMonth(String projectOid, String userId, Long startTime, Long endTime);
+
+    Long sumTotal(String projectOid, String userId, RechargeOrderStatus status, Long startTime, Long endTime);
+
 }

FullCardServer/src/main/java/com/zhongshu/card/server/core/dao/pay/impl/BalanceRechargeOrderDaoImpl.java

@@ -1,18 +1,26 @@
 package com.zhongshu.card.server.core.dao.pay.impl;
 
+import ch.qos.logback.core.util.StringUtil;
+import com.github.microservice.types.OrderState;
 import com.zhongshu.card.client.model.pay.QueryRechargeParam;
+import com.zhongshu.card.client.type.payAccount.RechargeOrderStatus;
 import com.zhongshu.card.client.utils.DateUtils;
 import com.zhongshu.card.server.core.dao.pay.extend.BalanceRechargeOrderDaoExtend;
 import com.zhongshu.card.server.core.domain.pay.BalanceRechargeOrder;
 import com.zhongshu.card.server.core.util.CommonUtil;
+import com.zhongshu.payment.client.model.TotalStatisticsModel;
+import com.zhongshu.payment.client.types.TradeType;
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 
+import java.util.ArrayList;
 import java.util.List;
 
 public class BalanceRechargeOrderDaoImpl implements BalanceRechargeOrderDaoExtend {
@@ -45,4 +53,48 @@ public class BalanceRechargeOrderDaoImpl implements BalanceRechargeOrderDaoExten
 
         return mongoTemplate.find(query, BalanceRechargeOrder.class);
     }
+
+    @Override
+    public Long sumTotal(String projectOid, String userId, RechargeOrderStatus status, Long startTime, Long endTime) {
+        Criteria criteria = new  Criteria();
+
+        if (StringUtils.isNotEmpty(projectOid)){
+            criteria.and("projectOid").is(projectOid);
+        }
+
+        if (StringUtils.isNotEmpty(userId)){
+            criteria.and("userId").is(userId);
+        }
+
+        if (status != null){
+            criteria.and("status").is(status);
+        }
+
+        if (!CommonUtil.longIsEmpty(startTime)){
+            criteria.and("createTime").gte(startTime);
+        }
+
+        if (CommonUtil.longIsEmpty(endTime)){
+            criteria.and("createTime").lte(endTime);
+        }
+
+        Aggregation aggregation = Aggregation.newAggregation(
+                Aggregation.match(criteria),
+//                Aggregation.project("_id","amountTotal"),
+                Aggregation.group("projectId", "userId").sum("total").as("totalAmount")
+//                Aggregation.group("tradeType").last("tradeType").as("_id")
+        );
+
+
+        AggregationResults<TotalStatisticsModel> groupList = mongoTemplate.aggregate(aggregation, BalanceRechargeOrder.class, TotalStatisticsModel.class);
+        List<TotalStatisticsModel> list = new ArrayList<>();
+
+        if (groupList.getMappedResults().size()==0){
+            return 0L;
+        }
+
+        return groupList.getMappedResults().get(0).getTotalAmount().longValue();
+    }
+
+
 }

FullCardServer/src/main/java/com/zhongshu/card/server/core/domain/pay/BalanceRechargeOrder.java

@@ -41,4 +41,7 @@ public class BalanceRechargeOrder extends SuperEntity {
 
     @Schema(description = "订单号")
     private String orderNo;
+
+    @Schema(description = "描述")
+    private String description;
 }

FullCardServer/src/main/java/com/zhongshu/card/server/core/service/pay/BalancePayService.java

@@ -18,6 +18,7 @@ import com.github.microservice.pay.client.type.ledger.TransactionType;
 import com.github.microservice.types.payment.PaymentChannelType;
 import com.github.microservice.types.payment.PaymentType;
 import com.wechat.pay.java.service.payments.model.Transaction;
+import com.zhongshu.card.client.model.mqtt.SendMessageModel;
 import com.zhongshu.card.client.model.pay.PayAccountParam;
 import com.zhongshu.card.client.model.pay.QueryRechargeParam;
 import com.zhongshu.card.client.model.pay.RechargeOrderModel;
@@ -30,10 +31,12 @@ import com.zhongshu.card.server.core.domain.org.UserAccount;
 import com.zhongshu.card.server.core.domain.pay.BalanceRechargeOrder;
 import com.zhongshu.card.server.core.domain.pay.PayAccount;
 import com.zhongshu.card.server.core.service.base.SuperService;
+import com.zhongshu.card.server.core.service.mqtt.MqttServiceImpl;
 import com.zhongshu.card.server.core.service.paySetting.OrgPayAccountService;
 import com.zhongshu.card.server.core.service.paySetting.ProjectPaySettingServiceImpl;
 import com.zhongshu.card.server.core.util.BeanUtils;
 import com.zhongshu.card.server.core.util.CommonUtil;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -46,6 +49,7 @@ import java.util.Map;
 import java.util.stream.Collectors;
 
 @Service
+@Slf4j
 public class BalancePayService extends SuperService {
 
     @Autowired
@@ -69,13 +73,16 @@ public class BalancePayService extends SuperService {
     @Autowired
     GeneralLedgerService generalLedgerService;
 
+    @Autowired
+    MqttServiceImpl mqttService;
+
 
     /**
      * 下单并生成调用支付参数
      */
     @Transactional
     public Object recharge(RechargeParam param) {
-        String projectOid = getCurrentOid();
+        String projectOid = getCurrentProjectOid();
         String userId = getCurrentUserId();
         String appId = getCurrentAppId();
 
@@ -109,12 +116,16 @@ public class BalancePayService extends SuperService {
         balanceRechargeOrder.setOrderNo(orderNo);
         balanceRechargeOrder.setTotal(new BigDecimal(param.getTotal()));
         balanceRechargeOrder.setPrepay(prepay.getContent());
+        balanceRechargeOrder.setDescription(param.getDescription());
         rechargeOrderDao.save(balanceRechargeOrder);
         return ResultContent.buildContent(toModel(balanceRechargeOrder));
     }
 
     public void handleWeChatCallback(WeChatMiniAppPayRet ret) {
         BalanceRechargeOrder rechargeOrder = rechargeOrderDao.findTopByOrderNo(ret.getOutTradeNo());
+        if (!rechargeOrder.getStatus().equals(RechargeOrderStatus.Wait)){
+            return;
+        }
         switch (ret.getTradeState()) {
             case "SUCCESS" -> {
                 rechargeOrder.setStatus(RechargeOrderStatus.Success);
@@ -148,6 +159,15 @@ public class BalancePayService extends SuperService {
                 if (transfer.getState().equals(ResultState.Success)) {
                     //处理订单状态
                     rechargeOrder.setStatus(RechargeOrderStatus.Success);
+                    // 发送成功通知消息
+                    SendMessageModel sendMessageModel = new SendMessageModel();
+                    sendMessageModel.setMessage(JsonUtil.toJson(Map.of("status", RechargeOrderStatus.Success.name(), "orderNo", rechargeOrder.getOrderNo())));
+                    String userId = rechargeOrder.getUserId();
+                    String topic = String.format("/wechat/pay/%s", userId);
+                    sendMessageModel.setTopic(topic);
+                    sendMessageModel.setUserId(userId);
+                    log.info("mq topic : {}", topic);
+                    mqttService.sendMessage(sendMessageModel);
                 }
             }
             case "PAYERROR", "REFUND" -> {
@@ -206,7 +226,8 @@ public class BalancePayService extends SuperService {
         Long monthStartTime = DateUtils.getMonthStartTime(param.getYear(), param.getMonth());
         Long monthEndTime = DateUtils.getMonthEndTime(param.getYear(), param.getMonth());
         List<BalanceRechargeOrder> balanceRechargeOrders = rechargeOrderDao.listByMonth(param.getProjectOid(), param.getUserId(), monthStartTime, monthEndTime);
-        return ResultContent.buildContent(balanceRechargeOrders.stream().map(this::toModel).toList());
+        Long amountTotal = rechargeOrderDao.sumTotal(param.getProjectOid(), param.getUserId(), RechargeOrderStatus.Success, monthStartTime, monthEndTime);
+        return ResultContent.buildContent(Map.of("list", balanceRechargeOrders.stream().map(this::toModel).toList(), "amountTotal", amountTotal));
     }
 
     public ResultContent<Long> queryUserBalance(String projectOid, String userId) {
@@ -226,6 +247,64 @@ public class BalancePayService extends SuperService {
         return ResultContent.build(ResultState.Fail);
     }
 
+    public ResultContent queryInfo(String orderNo){
+        BalanceRechargeOrder rechargeOrder = rechargeOrderDao.findTopByOrderNo(orderNo);
+        return ResultContent.buildContent(toModel(rechargeOrder));
+    }
+
+    public Object close(String orderNo){
+        BalanceRechargeOrder rechargeOrder = rechargeOrderDao.findTopByOrderNo(orderNo);
+        if (rechargeOrder==null){
+            return com.github.microservice.net.ResultContent.buildFail("订单号不存在");
+        }
+
+        if (!rechargeOrder.getStatus().equals(RechargeOrderStatus.Wait)){
+            return com.github.microservice.net.ResultContent.buildFail("当前状态不可关闭");
+        }
+
+        PayProductParameter<Object> parameter = new PayProductParameter<>();
+        String accountName = orgPayAccountService.queryOgPayAccount(rechargeOrder.getProjectOid(), rechargeOrder.getPaymentType());
+        parameter.setAccountName(accountName);
+        parameter.setMeta(Map.of("orderNo", orderNo));
+        ResultContent<Object> closeRet = miniAppPayService.close(parameter);
+        if (!closeRet.getState().equals(ResultState.Success)){
+            return com.github.microservice.net.ResultContent.buildFail("关闭失败");
+        }
+        rechargeOrder.setStatus(RechargeOrderStatus.Cancel);
+        rechargeOrderDao.save(rechargeOrder);
+        return ResultContent.build(ResultState.Success);
+    }
+
+    public ResultContent refund(String projectOid, String oid, String userId, String orderNo, BigDecimal total, String remark){
+        //获取用户余额支付子账户
+        PayAccount userAccount = payAccountService.getUserChildren(projectOid, userId, PaymentChannelType.BalancePayment);
+        //获取机构待结算子账户
+        PayAccount orgAccount = payAccountService.getOrgChildren(projectOid, oid, PaymentChannelType.WaitSettle);
+        //支付：用户余额子账+ 机构待结算子账-
+        //构建转账参数
+        TransferTransactionsModel transferModel = new TransferTransactionsModel();
+        //构建出账账户
+        TransferTransactionsModel.GeneralLedgerTransaction sourceTransaction = new TransferTransactionsModel.GeneralLedgerTransaction();
+        sourceTransaction.setAmount(total.negate().longValue());
+        sourceTransaction.setTransactionType(TransactionType.Refund);
+        sourceTransaction.setTransactionStatus(TransactionStatus.Success);
+        sourceTransaction.setOrderNumber(orderNo);
+        sourceTransaction.setGeneralLedgerId(orgAccount.getLedgerId());
+        sourceTransaction.setRemark(remark);
+        transferModel.setSource(new TransferTransactionsModel.GeneralLedgerTransaction[]{sourceTransaction});
+        //构建入账账户
+        TransferTransactionsModel.GeneralLedgerTransaction destinationTransaction = new TransferTransactionsModel.GeneralLedgerTransaction();
+        destinationTransaction.setAmount(total.longValue());
+        destinationTransaction.setTransactionType(TransactionType.Refund);
+        destinationTransaction.setTransactionStatus(TransactionStatus.Success);
+        destinationTransaction.setOrderNumber(orderNo);
+        destinationTransaction.setGeneralLedgerId(userAccount.getLedgerId());
+        destinationTransaction.setRemark(remark);
+        transferModel.setDestinations(new TransferTransactionsModel.GeneralLedgerTransaction[]{destinationTransaction});
+        //调用支付中心转账接口
+        return transactionLogService.transfer(transferModel);
+    }
+
     private RechargeOrderModel toModel(BalanceRechargeOrder balanceRechargeOrder) {
         RechargeOrderModel rechargeOrderModel = new RechargeOrderModel();
         if (balanceRechargeOrder != null) {

FullCardServer/src/main/java/com/zhongshu/card/server/core/stream/WeChatMiniAppPayStream.java

@@ -8,7 +8,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Slf4j
-@Component("WeChatMiniAppPayStream")
+@Component("weChatMiniAppPayStreamConsumer")
 public class WeChatMiniAppPayStream extends StreamConsumer<WeChatMiniAppPayRet> {
 
     @Autowired