package com.zanxiang.game.back.serve.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zanxiang.game.back.serve.dao.mapper.GameOceanengineAppCallbackMapper;
import com.zanxiang.game.back.serve.pojo.dto.GameOceanengineAppCallbackDTO;
import com.zanxiang.game.back.serve.pojo.entity.GameOceanengineAppCallback;
import com.zanxiang.game.back.serve.service.IGameOceanengineAppCallbackService;
import com.zanxiang.game.module.base.ServerInfo;
import com.zanxiang.game.module.base.pojo.vo.AgentRpcVO;
import com.zanxiang.game.module.base.rpc.IAgentRpc;
import com.zanxiang.module.util.DateUtil;
import com.zanxiang.module.util.JsonUtil;
import com.zanxiang.module.util.bean.BeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDate;

@Slf4j
@Service
public class GameOceanengineAppCallbackServiceImpl extends ServiceImpl<GameOceanengineAppCallbackMapper, GameOceanengineAppCallback>
        implements IGameOceanengineAppCallbackService {

    @Value("${spring.kafka.oceanengineAppCallbackTopic}")
    private String oceanengineAppCallbackTopic;
    @Autowired
    private KafkaProducer<String, String> kafkaProducer;
    @DubboReference(providedBy = ServerInfo.SERVER_DUBBO_NAME)
    private IAgentRpc agentRpc;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean callback(GameOceanengineAppCallbackDTO dto) {
        AgentRpcVO agent = agentRpc.getByByteAccountId(dto.getAdvertiserId()).getData();
        if (agent == null) {
            log.error("头条检测链接数据找不到渠道：{}", JsonUtil.toString(dto));
        }
        GameOceanengineAppCallback appCallback = BeanUtil.copy(dto, GameOceanengineAppCallback.class);
        if (agent != null) {
            appCallback.setAgentKey(agent.getAgentKey());
            appCallback.setGameId(agent.getGameId());
        } else {
            appCallback.setAgentKey("-");
            appCallback.setGameId(-1L);
        }
        appCallback.setDay(dto.getTs() == null || dto.getTs() < 1000 ? LocalDate.now() : DateUtil.milliToLocalDateTime(dto.getTs()).toLocalDate());
        save(appCallback);
        try {
            kafkaProducer.send(new ProducerRecord<>(oceanengineAppCallbackTopic, dto.getAdvertiserId().toString(), JsonUtil.toString(appCallback)));
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return true;
    }
}
