package com.zanxiang.game.module.sdk.service.impl;

import com.zanxiang.game.module.mybatis.entity.GameUserRole;
import com.zanxiang.game.module.sdk.enums.KafkaEventTrackEnum;
import com.zanxiang.game.module.sdk.pojo.param.GameRoleActiveCallParam;
import com.zanxiang.game.module.sdk.pojo.param.UserData;
import com.zanxiang.game.module.sdk.service.IKafkaService;
import com.zanxiang.module.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * @author : lingfeng
 * @time : 2024-03-07
 * @description : 卡夫卡队列
 */
@Slf4j
@Service
public class KafkaServiceImpl implements IKafkaService {

    @Value("${spring.kafka.game-sdk.gameRoleActiveTopic}")
    private String gameRoleActiveTopic;

    @Value("${spring.kafka.game-sdk.gameEventTrackTopic}")
    private String gameEventTrackTopic;

    @Autowired
    @Qualifier("gameSdkKafkaProducer")
    private Producer<String, String> kafkaProducer;

    @Override
    public void eventTrack(KafkaEventTrackEnum eventTrackEnum, String data) {
        try {
            kafkaProducer.send(new ProducerRecord<>(this.gameEventTrackTopic, eventTrackEnum.getValue(),
                    eventTrackEnum.getValue() + "=" + data));
        } catch (Exception e) {
            log.error("玩家数据埋点发送到 Kafka 异常!, eventTrackTopic : {}, eventTrackEnum : {}, data : {}, e : {}",
                    this.gameEventTrackTopic, eventTrackEnum.getValue(), data, e.getMessage());
        }
    }

    @Override
    public void roleActiveTrack(UserData userData, GameRoleActiveCallParam param) {
        //活跃提交
        Map<String, Object> activeParamMap = new HashMap<>(6);
        activeParamMap.put("userId", userData.getUserId());
        activeParamMap.put("gameId", userData.getGameId());
        activeParamMap.put("serverId", param.getServerId());
        activeParamMap.put("roleId", param.getRoleId());
        activeParamMap.put("roleLevel", param.getRoleLevel());
        activeParamMap.put("activeTime", System.currentTimeMillis());
        try {
            kafkaProducer.send(new ProducerRecord<>(this.gameRoleActiveTopic, userData.getUserId().toString(), JsonUtil.toString(activeParamMap)));
        } catch (Exception e) {
            log.error("玩家角色活跃信息发送到 Kafka 异常!, eventTrackTopic : {}, data : {}, e : {}",
                    this.gameRoleActiveTopic, JsonUtil.toString(activeParamMap), e.getMessage());
        }
    }

    @Override
    public void roleActiveTrack(GameUserRole gameUserRole, Integer activeType) {
        //活跃提交
        Map<String, Object> activeParamMap = new HashMap<>(6);
        activeParamMap.put("userId", gameUserRole.getUserId());
        activeParamMap.put("gameId", gameUserRole.getGameId());
        activeParamMap.put("roleId", gameUserRole.getRoleId());
        activeParamMap.put("serverId", gameUserRole.getServerId());
        activeParamMap.put("roleLevel", gameUserRole.getRoleLevel());
        activeParamMap.put("activeType", activeType);
        activeParamMap.put("activeTime", System.currentTimeMillis());
        try {
            kafkaProducer.send(new ProducerRecord<>(this.gameRoleActiveTopic, gameUserRole.getUserId().toString(), JsonUtil.toString(activeParamMap)));
        } catch (Exception e) {
            log.error("玩家角色活上下线跃信息发送到 Kafka 异常!, eventTrackTopic : {}, data : {}, e : {}",
                    this.gameRoleActiveTopic, JsonUtil.toString(activeParamMap), e.getMessage());
        }
    }
}
