package ru.micord.ervu.account_applications.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import ru.micord.ervu.account_applications.kafka.model.DeclarationStatus;
import ru.micord.ervu.account_applications.security.model.jwt.UserClaims;
import ru.micord.ervu.account_applications.security.service.EncryptionService;
import ru.micord.ervu.account_applications.security.service.JwtTokenService;
import ru.micord.ervu.account_applications.service.UserApplicationListService;

/**
 * @author Adel Kalimullin
 */
@Component
public class DeclarationStatusListener {
  private static final Logger LOGGER = LoggerFactory.getLogger(DeclarationStatusListener.class);
  private final ObjectMapper mapper;
  private final UserApplicationListService applicationService;
  private final EncryptionService encryptionService;
  private final JwtTokenService jwtTokenService;

  public DeclarationStatusListener(ObjectMapper mapper,
      UserApplicationListService applicationService, EncryptionService encryptionService,
      JwtTokenService jwtTokenService) {
    this.mapper = mapper;
    this.applicationService = applicationService;
    this.encryptionService = encryptionService;
    this.jwtTokenService = jwtTokenService;
  }

  @KafkaListener(id = "${kafka.application.status.group.id}",
      topics = "${kafka.application.status}")
  public void listenKafkaDomain(String kafkaMessage, @Header("authorization") String authHeader) {
    try {
      String token = authHeader.replace("Bearer ", "");
      UserClaims userClaims = jwtTokenService.getUserClaims(token);
      String name = userClaims.name();
      String userId = userClaims.userId();

      DeclarationStatus declarationStatus = mapper.readValue(kafkaMessage, DeclarationStatus.class);
      Long applicationNumber = declarationStatus.applicationNumber();
      if (declarationStatus.status()) {
        LOGGER.info("update by appNumber = {}", applicationNumber);
        String tempPass = declarationStatus.password();
        if (StringUtils.hasText(tempPass)) {
          String encryptedPassword = encryptionService.encrypt(tempPass);
          applicationService.savePassword(applicationNumber, encryptedPassword, name, userId);
        }
        else {
          applicationService.saveAcceptedStatus(applicationNumber, name, userId);
        }
      }
      else {
        String errorMsg = declarationStatus.errorMsg();
        LOGGER.error("error by appNumber = {}, message: {}", applicationNumber, errorMsg);
        applicationService.saveError(applicationNumber, errorMsg, name, userId);
      }
    }
    catch (JsonProcessingException e) {
      LOGGER.error("Failed to deserialize message: {}", kafkaMessage, e);
    }
  }
}
