package com.gwm.marketing.kafka.product;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.gwm.marketing.dto.user.UserSyncIdentificationDto;
import org.apache.kafka.common.protocol.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author fanht
 * @descrpiton
 * @date 2022/10/21 15:23:02
 * @versio 1.0
 */
@Component
public class KafkaProductDemo {


    private Logger logger = LoggerFactory.getLogger(this.getClass());
    @Value("${spring.kafka.consumer.syncUserIdentify}")
    private String kafkaTopic;

    @Resource
    private KafkaTemplate kafkaTemplate;

    public void sendKafkaMessage(){
        Gson gson = new Gson();
        UserSyncIdentificationDto dto = UserSyncIdentificationDto.builder().name("官方")
                .beanId("3032395126028730368").identifyId("3032395126028730368").typeCode("OFFICIAL").updateTime(System.currentTimeMillis()).build();
        UserSyncIdentificationDto dto1 = UserSyncIdentificationDto.builder().identifyId("3032395126028730368").updateTime(System.currentTimeMillis())
                        .beanId("3032395126028730368").typeCode("OFFICIAL").build();
        System.out.println("入參：" + gson.toJson(dto));
        logger.info("請求入參:" + gson.toJson(dto));
        kafkaTemplate.send(kafkaTopic,null, "1",gson.toJson(dto));
        kafkaTemplate.send(kafkaTopic,null, "2", gson.toJson(dto));
    }

}

package com.gwm.marketing.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.gwm.marketing.common.constants.UserConstants;
import com.gwm.marketing.common.dto.user.CancleUser;
import com.gwm.marketing.common.enums.TypeCodeEnum;
import com.gwm.marketing.constants.CommonConstants;
import com.gwm.marketing.dao.user.UserSyncIdentificationDao;
import com.gwm.marketing.dto.user.UserSyncIdentificationDto;
import com.gwm.marketing.entity.user.UserSyncIdentification;
import com.gwm.marketing.feign.community.FeignCommunityClient;
import com.gwm.marketing.service.user.UserDetailService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.Optional;

/**
 * @author fanht
 * @descrpiton 消費仙豆kafka消息
 * @date 2022/10/21 14:41:59
 * @versio 1.0
 */
@Component
public class UserSyncIdentificationConsumer {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private UserDetailService userDetailService;

    @Resource
    private UserSyncIdentificationDao userSyncIdentificationDao;

    @Resource
    FeignCommunityClient feignCommunityClient;



    Gson gson = new GsonBuilder().create();

    @KafkaListener(topics = "${spring.kafka.consumer.syncUserIdentify}", topicPartitions = {}, groupId = "test")
    public void onMessage(ConsumerRecord<?, ?> record) {
        String key = (String) record.key();
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent() && StringUtils.isNotEmpty(key)) {
            try {
                logger.info("===仙豆入參==" +JSONObject.toJSONString(kafkaMessage));
                Object message = kafkaMessage.get();

                switch (key) {
                    case CommonConstants.SAVE:
                        //導入、添加、or修改
                        updateOrCreateIdentify(message);
                        break;
                    case CommonConstants.CANCEL:
                        //撤銷認證
                        cancleUserIdentification(message);
                        break;
                    default:
                        break;
                }

            } catch (Exception e) {
                logger.error("=====創建異常======", e);
            }
        }
    }


    private void updateOrCreateIdentify(Object message) {

        logger.debug("====仙豆入參====" + JSONObject.toJSONString(message));
        //UserSyncIdentificationDto dto = gson.fromJson(message.toString(), UserSyncIdentificationDto.class);
        UserSyncIdentificationDto dto = JSONObject.toJavaObject(JSON.parseObject(message.toString()), UserSyncIdentificationDto.class);
        logger.info("======javabean轉化對象成功======" + JSONObject.toJSONString(dto));
        String userId = userDetailService.getUserIdByBeanId(dto.getBeanId(), CommonConstants.ORA);
        UserSyncIdentification usi = UserSyncIdentification.builder().userId(userId).typeCode(dto.getTypeCode())
                .createTime(new Date()).beanId(dto.getBeanId()).extra1(dto.getIdentifyId())
                .name(dto.getName()).sourceApp(CommonConstants.ORA).extra2(dto.getUpdateTime()==null?null:dto.getUpdateTime().toString()).build();
        if (dto.getId() != null && dto.getId() > 0) {
            userSyncIdentificationDao.updateByPrimaryKeySelective(usi);
        } else {
            //todo 此處可能會出現多次插入，是否需要做限制
           int count = userSyncIdentificationDao.countByBeanIdAndExtra1(usi.getBeanId(),usi.getExtra1());
           if(count <= 0){
               userSyncIdentificationDao.insertSelective(usi);
               try {
                   logger.info("==========更新同步es===start" + JSONObject.toJSONString(dto));
                   this.syncEs(dto);
                   logger.info("==========更新同步es===end");
               } catch (Exception e) {
                   logger.error("同步es異常",e);
               }
           }else {
               logger.info("========數據重復==,入參：" +JSONObject.toJSONString(usi));
           }
        }
    }

    private void cancleUserIdentification(Object message) {
        logger.debug("====仙豆取消入參====" + JSONObject.toJSONString(message));
        //UserSyncIdentificationDto dto = gson.fromJson(message.toString(), UserSyncIdentificationDto.class);
        UserSyncIdentificationDto dto = JSONObject.toJavaObject(JSON.parseObject(message.toString()), UserSyncIdentificationDto.class);
        logger.info("======javabean轉化對象成功======" + JSONObject.toJSONString(dto));
        int cancleResult = userSyncIdentificationDao.cancleIdentifies(dto.getUpdateTime().toString(),dto.getBeanId(), dto.getIdentifyId());
        if(cancleResult <= 0){
            logger.info("====用戶撤銷認證同步失敗======" +JSONObject.toJSONString(dto));
        }else {
            //操作es，若當前撤銷認證的用戶是歐拉用戶，且身份是官方且已佩戴，則更改es中的帖子對應的官方身份狀態(佩戴或者不佩戴都更改es)
            try {
                logger.info("==========取消同步es===start" + JSONObject.toJSONString(dto));
                this.syncEs(dto);
                logger.info("==========取消同步es===end");
            } catch (Exception e) {
                logger.error("更新es官方數據失敗",  e);
            }
        }
    }

    public void syncEs(UserSyncIdentificationDto dto){
        if(TypeCodeEnum.OFFICIAL.getCode().equals(dto.getTypeCode())){
            String userId = userDetailService.getUserIdByBeanId(dto.getBeanId(), CommonConstants.ORA);
            CancleUser cancleUser = new CancleUser();
            cancleUser.setIsOfficial(UserConstants.IS_OFFICIAL_NO);
            cancleUser.setCreateBy(userId);
            cancleUser.setSourceApp(CommonConstants.ORA);
            feignCommunityClient.updateEsUserStatus(cancleUser);
            logger.debug("========更改es成功======");
        }
    }
}