【DAOP-1.0】企业认证

【功能点】kafka方式改造

 package com.csbr.qingcloud.portal.controller;
 
 import com.csbr.cloud.common.response.CommonRes;
+import com.csbr.cloud.workflow.domain.dto.appove.FlowUpdateStateDTO;
 import com.csbr.qingcloud.portal.domain.vo.*;
 import com.csbr.qingcloud.portal.service.EnterpriseChangeApproveService;
 import csbr.cloud.entity.annotation.SystemLog;
@@ -93,6 +94,14 @@ public class EnterpriseController {
         return CommonRes.success(true);
     }
 
+    @PostMapping("/last-approve")
+    @SystemLog(value = "企业认证审批")
+    @Operation(summary = "企业认证审批-最后一个节点时调用")
+    public CommonRes<Boolean> lastApprove(@RequestBody FlowUpdateStateDTO dto) {
+        Boolean vo = enterpriseService.lastApprove(dto);
+        return CommonRes.success(vo);
+    }
+
     @PostMapping("/move-old-data")
     @SystemLog(value = "将此账号提交的数据需求迁移到新的企业")
     @Operation(summary = "将此账号提交的数据需求迁移到新的企业", hidden = true)

 package com.csbr.qingcloud.portal.service;
 
+import com.csbr.cloud.workflow.domain.dto.appove.FlowUpdateStateDTO;
 import com.csbr.qingcloud.portal.domain.vo.*;
 import csbr.cloud.entity.domain.base.vo.PageListVO;
 import com.csbr.cloud.workflow.domain.dto.callback.BizCallbackDTO;
@@ -77,4 +78,6 @@ public interface EnterpriseService {
 
     /** 将此账号提交的数据需求迁移到新的企业 */
     void moveOldData(TenantRQVO vo);
+
+    Boolean lastApprove(FlowUpdateStateDTO dto);
 }

 package com.csbr.qingcloud.portal.service.impl;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.csbr.cloud.common.enums.SystemError;
 import com.csbr.cloud.common.enums.UserError;
 import com.csbr.cloud.common.enums.WorkFlowBizEnum;
 import com.csbr.cloud.common.exception.CsbrUserException;
+import com.csbr.cloud.mq.service.KafkaStdService;
+import com.csbr.cloud.workflow.domain.dto.appove.FlowUpdateStateDTO;
+import com.csbr.cloud.workflow.feign.WorkflowFeign;
 import com.csbr.qingcloud.portal.domain.vo.*;
 import com.csbr.qingcloud.portal.feign.PersonelFeign;
 import com.csbr.qingcloud.portal.feign.UserFeign;
@@ -16,6 +20,7 @@ import com.csbr.qingcloud.portal.mybatis.service.MfDemandService;
 import com.csbr.qingcloud.portal.mybatis.service.MfEnterpriseChangeApproveService;
 import com.csbr.qingcloud.portal.service.EnterpriseChangeApproveService;
 import com.csbr.qingcloud.portal.util.ObjectUtil;
+import csbr.cloud.entity.domain.base.dao.BaseShardingDO;
 import csbr.cloud.entity.enums.ApprovalStateEnum;
 import com.csbr.cloud.common.exception.CsbrSystemException;
 import com.csbr.cloud.common.util.CommonUtil;
@@ -30,11 +35,16 @@ import com.csbr.cloud.workflow.domain.dto.callback.BizCallbackDTO;
 import com.csbr.qingcloud.portal.mybatis.entity.MfEnterprise;
 import com.csbr.qingcloud.portal.mybatis.service.MfEnterpriseService;
 import com.csbr.qingcloud.portal.service.EnterpriseService;
+import csbr.cloud.entity.utils.UserContextHolder;
 import io.seata.common.util.DateUtil;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Service;
 import io.seata.spring.annotation.GlobalTransactional;
 import org.springframework.transaction.annotation.Transactional;
@@ -82,6 +92,12 @@ public class EnterpriseServiceImpl extends FlowAbstractImpl implements Enterpris
     @Resource
     private UserFeign userFeign;
 
+    @Resource
+    private WorkflowFeign workflowFeign;
+
+    @Autowired(required = false)
+    private KafkaStdService kafkaStdService;
+
     /**
      * 企业信息分页查询
      * @author xcq
@@ -387,6 +403,33 @@ public class EnterpriseServiceImpl extends FlowAbstractImpl implements Enterpris
         approvalFlowUtil.removeApproveByBizGuids(guids);
     }
 
+    /** ***企业认证*** */
+    @GlobalTransactional
+    @Override
+    public Boolean lastApprove(FlowUpdateStateDTO dto) {
+        //如果是非认证会员认证审批通过了 则需要走kafka生成新会员
+        dto.setBackEndInvoke(true);
+        String topic = "tenant.enterprise";
+        kafkaStdService.stdSendMsgWithPrefix(topic, JSON.toJSONString(dto));
+        return true;
+    }
+
+    @KafkaListener(containerFactory = "kafkaListenerContainerFactory", groupId = "scm.notify", topics = "${csbr.kafka.topic-prefix:}_"+"approve.tenant.enterprise",
+            errorHandler = "consumerAwareErrorHandler")
+    public void addUserListener(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
+        ack.acknowledge();
+        for (ConsumerRecord<String, String> record : records) {
+            try {
+                FlowUpdateStateDTO dto = JSON.parseObject(record.value(),FlowUpdateStateDTO.class);
+                workflowFeign.allowFlowCore(dto).getData();
+            } catch (Exception e) {
+                log.error("企业认证出现异常: " + e.getMessage() + ";\nRecord Info: " + record.toString());
+                e.printStackTrace();
+                return;
+            }
+        }
+    }
+    
     /**
      * 流程结束后进行业务回调
      * @author xcq
@@ -422,7 +465,7 @@ public class EnterpriseServiceImpl extends FlowAbstractImpl implements Enterpris
 
     /** 将此账号提交的数据需求迁移到新的企业 */
     @Override
-    @Transactional
+    @Transactional(rollbackFor = Exception.class)
     public void moveOldData(TenantRQVO vo){
         LambdaQueryWrapper<MfDemand> queryWrapper = new LambdaQueryWrapper<>();
         queryWrapper.eq(MfDemand::getRequirementUnit, vo.getTenantName());