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

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zanxiang.game.back.serve.dao.mapper.GameOceanengineCallbackMapper;
import com.zanxiang.game.back.serve.pojo.dto.GameOceanengineCallbackDTO;
import com.zanxiang.game.back.serve.pojo.entity.GameOceanengineCallback;
import com.zanxiang.game.back.serve.service.IGameOceanengineCallbackService;
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 java.time.LocalDate;

@Slf4j
@Service
public class GameOceanengineCallbackServiceImpl extends ServiceImpl<GameOceanengineCallbackMapper, GameOceanengineCallback>
        implements IGameOceanengineCallbackService {

    @Value("${spring.kafka.oceanengineCallbackTopic}")
    private String oceanengineCallbackTopic;

    @DubboReference(providedBy = ServerInfo.SERVER_DUBBO_NAME)
    private IAgentRpc agentRpc;

    @Autowired
    private KafkaProducer<String, String> kafkaProducer;

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