Untitled

package com.muf.integrasi_ppd_java.config;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.muf.integrasi_ppd_java.dto.ContractDTO;
import com.muf.integrasi_ppd_java.dto.LogidDTO;
import com.muf.integrasi_ppd_java.dto.ParamOutDTO;
import com.muf.integrasi_ppd_java.helper.HelpersAM;
import com.muf.integrasi_ppd_java.helper.NullValueHelper;
import com.muf.integrasi_ppd_java.model.mufam.FlaggingIntegration;
import com.muf.integrasi_ppd_java.repository.mufam.ContractMasterRepository;
import com.muf.integrasi_ppd_java.repository.mufam.FlaggingIntegrationRepository;
import com.muf.integrasi_ppd_java.repository.mufam.MstDomainDtlRepository;
import com.muf.integrasi_ppd_java.repository.procedure.CallingProcedureRepository;
import com.muf.integrasi_ppd_java.service.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class Consumer {

    @Autowired
    private ContractBufferCashService contractBufferCashService;
    @Autowired
    private ApplicationsService applicationsService;
    @Autowired
    private ApplicationSalesForceInfoService applicationSalesForceInfoService;
    @Autowired
    private ApplicationSurveyService applicationSurveyService;
    @Autowired
    private ConfirmationLetterService confirmationLetterService;
    @Autowired
    private ContractInsuranceAdditionService contractInsuranceAdditionService;
    @Autowired
    private ContractInsuranceFeeService contractInsuranceFeeService;
    @Autowired
    private ContractLoanFeeService contractLoanFeeService;
    @Autowired
    private ContractLoanStructureService contractLoanStructureService;
    @Autowired
    private ContractLoanRefundService contractLoanRefundService;
    @Autowired
    private ContractLoanSubsidyService contractLoanSubsidyService;
    @Autowired
    private ContractMasterService contractMasterService;
    @Autowired
    private ContractObjectService contractObjectService;
    @Autowired
    private ContractProcessTimeService contractProcessTimeService;
    @Autowired
    private CustomerAddressDetailService customerAddressDetailService;
    @Autowired
    private CustomerCreditCardService customerCreditCardService;
    @Autowired
    private CustomerGuarantorService customerGuarantorService;
    @Autowired
    private CustomerMasterService customerMasterService;
    @Autowired
    private CustomerPersonalEmployeesService customerPersonalEmployeesService;
    @Autowired
    private CustomerPersonalProfessionalService customerPersonalProfessionalService;
    @Autowired
    private CustomerPersonalService customerPersonalService;
    @Autowired
    private CustomerPropertyDetailService customerPropertyDetailService;
    @Autowired
    private PurchaseOrderService purchaseOrderService;
    @Autowired
    private CustomerBankAccount2Service customerBankAccountService;
    @Autowired
    private CustomerPersonalEnterpriseService customerPersonalEnterpriseService;
    @Autowired
    private FlaggingIntegrationRepository flaggingIntegrationRepository;
    @Autowired
    private CustomerPersonalFamilyService customerPersonalFamilyService;
    @Autowired
    private ContractMasterRepository contractMasterRepository;
    @Autowired
    private CustomerPersonalOccupationService customerPersonalOccupationService;
    @Autowired
    private ContractAmortizeService contractAmortizeService;
    @Autowired
    private ContractDisburseFeeService contractDisburseFeeService;
    @Autowired
    private ContractObjectExtendedService contractObjectExtendedService;
    @Autowired
    private AmBastkInvoiceService amBastkInvoiceService;
    @Autowired
    private CallingProcedure callingProcedure;
    @Autowired
    private MstDomainDtlRepository domainDtlRepository;
    @Autowired
    private CallingProcedureRepository callingProcedureRepository;
    @Autowired
    private ApplicationDeviasiService applicationDeviasiService;
    @Autowired
    private ApplicationVertelHeaderService applicationVertelHeaderService;
    @Autowired
    private ApplicationVertelHistoryService applicationVertelHistoryService;
    @Autowired
    private ApplicationVertelOtsService applicationVertelOtsService;
    @Autowired
    private DocReceiveHeaderService docReceiveHeaderService;
    @Autowired
    private ApplicationApprovalReturnService applicationApprovalReturnService;
    @Autowired
    private ApplicationScoringService applicationScoringService;
    @Autowired
    private ApplicationScoringDetailService applicationScoringDetailService;
    @Autowired
    private CreditApprovalHeaderService creditApprovalHeaderService;

    public static final Integer FLAG_INITIAL = 0;
    public static final Integer FLAG_SUCCESS = 1;
    public static final Integer FLAG_FAILURE = 2;
    public static final Integer FLAG_RETRY = 5;
    public static final Integer FLAG_MAPPING = 4;


    //private ExecutorService executor = Executors.newSingleThreadExecutor();


    @KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group-id}")
    @Transactional
    public void consume(String param) throws Exception {
        ObjectMapper mapper = new ObjectMapper();
        Map<String, Object> json_data = mapper.readValue(param, Map.class);

        String contractNo = HelpersAM.parsingContractNo(json_data);
        saveToFlaggingTable(contractNo, FLAG_INITIAL, FLAG_INITIAL, "INITIAL", "INTEGRASI", new Date());

        Map<String, Object> detail = (Map<String, Object>) json_data.get("detail");
        Map<String, Object> dataEntryCompletion = (Map<String, Object>) detail.get("data_entry_completion");
        Map<String, Object> verificationData = (Map<String, Object>) detail.get("verification_data");
        Map<String, Object> prePpd = (Map<String, Object>) verificationData.get("pre_ppd");
        String createdBy = NullValueHelper.resultStrFromJson(dataEntryCompletion, "confirmed_by");
        Date createdDate = NullValueHelper.resultDateFromJson(dataEntryCompletion, "confirmed_date");
        String branchCode = NullValueHelper.resultStrFromJson(json_data, "branch_code");

        String tanggalCutOff = domainDtlRepository.findTanggalCutoff("PPD01", "INTEGRASI");
        String datePrePpd = NullValueHelper.resultStrFromJson(prePpd, "confirmation_date");
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd");
        Date tanggalCutOffConvert = dateFormat.parse(tanggalCutOff);
        Date tanggalPrePpd = dateFormat.parse(datePrePpd);
        LogidDTO logidDTO = callingProcedureRepository.generateLogid();

        if (tanggalCutOffConvert.before(tanggalPrePpd) || tanggalCutOffConvert.equals(tanggalPrePpd)) {
            try {
                System.out.println(logidDTO.getLogid());
                asyncHandleKafkaIntegration(contractNo, json_data, createdBy, createdDate, logidDTO);
            } catch (Exception e) {
                log.error("Error reading value from Kafka: {}", e.getMessage(), e);
                return;
            }
        }
        callingProcedure.insertLogs(logidDTO.getLogid(), createdBy, "INTEGRASI", "1", branchCode);
    }

    //@Async
    @Transactional
    public void asyncHandleKafkaIntegration(String contractNo, Map<String, Object> data, String createdBy, Date createdDate, LogidDTO logidDTO) throws Exception {
        try {
            handleKafkaIntegration(contractNo, data, createdBy, createdDate, logidDTO);
        } catch (Exception e) {
            log.error("Transaksi gagal di kafka integrasi: " + e.getMessage(), e);
            throw e; // pastikan exception dilempar kembali
        }
    }

    @Transactional
    public void handleKafkaIntegration(String contractNo, Map<String, Object> data, String createdBy, Date createdDate, LogidDTO logidDTO) throws Exception {
        try {
            //bridging data
            processKafkaIntegrasi(contractNo, data, createdBy, createdDate, logidDTO);

            //update data mapping param
            processKafkaParam(contractNo, data, createdBy, createdDate, logidDTO);

            // pembentukan jurnal & kp
            processKafkaJournal(contractNo, createdBy, createdDate);
        } catch (Exception e) {
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            throw e;
        }
    }

    @Transactional
    public void processKafkaIntegrasi(String contractNo, Map<String, Object> data, String createdBy, Date createdDate, LogidDTO logidDTO) throws Exception {
        try {
            String result = performCombinedInsertion(data, logidDTO.getLogid());
            System.out.println("message error: " + result);
            if ("Success".equals(result)) {
                log.info("Transaksi berhasil di bridging");
                saveToFlaggingTable(contractNo, FLAG_MAPPING, FLAG_INITIAL, "BERHASIL INTEGRASI", createdBy, createdDate);
            } else {
                log.error("process Kafka Integrasi - Kondisi ELSE Ditemukan: " + result);
                //saveToFlaggingTable(contractNo, FLAG_FAILURE, FLAG_INITIAL, result, createdBy, createdDate);
                //throw new Exception(result);
                //System.out.println("contract no: " + contractNo);
                //FlaggingIntegration dataLogs = flaggingIntegrationRepository.getByContractNo(contractNo).orElse(FlaggingIntegration.builder().logs("kosong data").build());
                //System.out.println("data logs: " + dataLogs);
            }
        } catch (Exception e) {
            //log.error("Transaksi gagal di bridging: " + e.getMessage(), e);
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            saveToFlaggingTable(contractNo, FLAG_FAILURE, FLAG_INITIAL, e.getMessage(), createdBy, createdDate);
            throw e;
        }
    }

    @Transactional
    public void processKafkaParam(String contractNo, Map<String, Object> data, String createdBy, Date createdDate, LogidDTO logidDTO) {
        try {
            String statusIntegrasi = cekStatusIntegrasi(contractNo);
            if (statusIntegrasi.equals("4")) {
                //String logid = contractMasterRepository.findLogidByContractNo(contractNo).orElse(null);
                ContractDTO contractDTO = contractMasterRepository.findByContractNo(contractNo).orElse(ContractDTO.builder().build());
                ParamOutDTO paramOutDTO = callingProcedureRepository.updateParam(contractDTO.getContractId());
                if (paramOutDTO.getPOut() == null) {
                    try {
                        switch (paramOutDTO.getPTypeId()) {
                            case "01":
                                customerPersonalProfessionalService.insertData(data, paramOutDTO.getPEconomicSector(), logidDTO.getLogid());
                                break;
                            case "02":
                                customerPersonalEnterpriseService.insertData(data, paramOutDTO.getPEconomicSector(), logidDTO.getLogid());
                                break;
                            default:
                                customerPersonalEmployeesService.insertData(data, paramOutDTO.getPEconomicSector(), logidDTO.getLogid());
                                break;
                        }
                        saveToFlaggingTable(contractNo, FLAG_SUCCESS, FLAG_INITIAL, "BERHASIL INTEGRASI DAN MAPPING PARAMETER", createdBy, createdDate);
                        log.info("Transaksi berhasil di update param");
                    } catch (Exception e) {
                        saveToFlaggingTable(contractNo, FLAG_MAPPING, FLAG_INITIAL, e.getMessage(), createdBy, createdDate);
                        throw new Exception(e);
                    }
                } else {
                    saveToFlaggingTable(contractNo, FLAG_MAPPING, FLAG_INITIAL, paramOutDTO.getPOut(), createdBy, createdDate);
                    throw new Exception(paramOutDTO.getPOut());
                }
            }
        } catch (Exception e) {
            log.error("Transaksi gagal di mapping parameter: " + e.getMessage(), e);
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
        }
    }

    @Transactional(rollbackFor = Exception.class)
    public void processKafkaJournal(String contractNo, String createdBy, Date createdDate) {
        try {
            String statusIntegrasi = cekStatusIntegrasi(contractNo);
            String statusJurnalKp = cekStatusJurnalKp(contractNo);

            if ("1".equals(statusIntegrasi) && "0".equals(statusJurnalKp)) {
                String validasi = callingProcedureRepository.journalPpd(contractNo);
                if (validasi == null || validasi.isEmpty()) {
                    saveToFlaggingTable(contractNo, FLAG_SUCCESS, FLAG_SUCCESS, "BERHASIL PEMBENTUKAN JURNAL KP", createdBy, createdDate);
                } else {
                    saveToFlaggingTable(contractNo, FLAG_SUCCESS, FLAG_FAILURE, validasi, createdBy, createdDate);
                    throw new RuntimeException(validasi);
                }
            }
        } catch (Exception e) {
            log.error("Transaksi gagal pembentukan jurnal KP: " + e.getMessage(), e);
            throw e;
        }
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = Exception.class)
    public void saveToFlaggingTable(String contractNo, Integer statusIntegrasi, Integer statusJurnalKp,
                                    String logs, String createdBy, Date createdDate) {
        FlaggingIntegration request = flaggingIntegrationRepository.getByContractNo(contractNo)
                .orElse(FlaggingIntegration.builder()
                        .contractNo(contractNo)
                        .createdBy(createdBy)
                        .build());
        request.setStatusIntegrasi(statusIntegrasi);
        request.setStatusJurnalKp(statusJurnalKp);
        if (request.getCreatedDate() == null) {
            request.setUpdatedDate(null);
            request.setUpdatedBy(null);
            request.setCreatedDate(createdDate);
        } else {
            request.setUpdatedBy(createdBy);
            request.setUpdatedDate(new Date());
        }
        if (logs.length() > 250) {
            logs = logs.substring(0, 250);
        }
        request.setLogs(logs);
        flaggingIntegrationRepository.save(request);
    }

    public String cekStatusIntegrasi(String contractNo) {
        return flaggingIntegrationRepository.getStatusIntegrasi(contractNo);
    }

    public String cekStatusJurnalKp(String contractNo) {
        return flaggingIntegrationRepository.getStatusJurnalKp(contractNo);
    }

    @Transactional
    public String performCombinedInsertion(Map<String, Object> json_data, String logsId) throws Exception {
        //try {
            try {
                System.out.println(json_data);
                String contractNo = HelpersAM.parsingContractNo(json_data);
                ContractDTO contractMaster = contractMasterRepository.findByContractNo(contractNo).orElse(null);
                //--------------------TABLE EXISTING--------------------
                applicationsService.updateData(json_data, contractMaster, logsId); //update
                applicationSurveyService.insertData(json_data, contractMaster, logsId);
                contractMasterService.updateData(json_data, contractMaster, logsId); //update
                contractObjectService.updateDate(json_data, contractMaster, logsId); //update
                contractInsuranceAdditionService.insertData(json_data, contractMaster);
                contractInsuranceFeeService.updateData(json_data, contractMaster, logsId); //update
                contractLoanFeeService.insertData(json_data, contractMaster, logsId);
                contractLoanRefundService.insertData(json_data, contractMaster, logsId);
                contractLoanStructureService.updateData(json_data, contractMaster, logsId); //update
                contractLoanSubsidyService.insertData(json_data, contractMaster, logsId);
                customerMasterService.updateDate(json_data, contractMaster); //update
                //customerCompanyService.insertData(json_data, contractMaster); //--SKIP (table untuk kebutuhan company / badan usaha)
                customerPersonalService.insertData(json_data, contractMaster);
                customerAddressDetailService.insertData(json_data, contractMaster);
                customerBankAccountService.insertData(json_data, contractMaster, logsId);
                //customerDocDetailService.insertData(json_data, contractMaster); //--SKIP (table untuk kebutuhan company / badan usaha)
                //customerShareholderService.insertData(json_data, contractMaster); //--SKIP (table untuk kebutuhan company / badan usaha)
                customerGuarantorService.insertData(json_data, contractMaster);
                //addressGuarantorService.insertData(json_data, contractMaster); //jadi satu dengan java customer address detail karena mengacu ke table yang sama
                customerPropertyDetailService.insertData(json_data, contractMaster);
                customerPersonalFamilyService.insertDataV2(json_data, contractMaster);
                customerCreditCardService.insertData(json_data, contractMaster);
                purchaseOrderService.updateData(json_data, contractMaster, logsId); //update
                confirmationLetterService.updateData(json_data, contractMaster, logsId); //update
                applicationSalesForceInfoService.insertData(json_data, contractMaster, logsId); //--SKIP TIDAK ADA SALES FORCE ID DI JSON (APPL SALES FORCE ID SEQUENCE DI ACQ)
                contractBufferCashService.insertData(json_data, contractMaster);
                contractProcessTimeService.insertData(json_data, contractMaster, logsId);
                //applicationLapKeuService.insertData(json_data, contractMaster); //--SKIP (table untuk kebutuhan company / badan usaha)
                customerPersonalService.saveDataMapping(json_data, contractMaster); // function untuk menyimpan data2 ke table staging param
                amBastkInvoiceService.insertData(json_data, contractMaster, logsId);
                //--------------------TABLE EXISTING--------------------

                //--------------------TABLE BARU------------------------
//                customerPersonalOccupationService.insertData(json_data, contractMaster, logsId);
//                contractAmortizeService.insertData(json_data, contractMaster, logsId);
//                contractDisburseFeeService.updateData(json_data, contractMaster, logsId);
//                contractObjectExtendedService.updateData(json_data, contractMaster, logsId);
//                applicationDeviasiService.insertData(json_data, contractMaster, logsId);
//                applicationVertelHeaderService.insertData(json_data, contractMaster, logsId);
//                applicationVertelHistoryService.insertData(json_data, contractMaster, logsId);
//                applicationVertelOtsService.insertData(json_data, contractMaster, logsId);
//                docReceiveHeaderService.insertData(json_data, contractMaster, logsId);
//                applicationApprovalReturnService.insertData(json_data, contractMaster, logsId);
//                applicationScoringService.insertData(json_data, contractMaster, logsId);
//                applicationScoringDetailService.insertData(json_data, contractMaster, logsId);
//                creditApprovalHeaderService.insertData(json_data, contractMaster, logsId);
                //--------------------TABLE BARU-------------------------
                return "Success";
            } catch (Exception e) {
                //String errorMessage = "Error Proses Bridging Pada Bagian: " + e.getMessage() + " Pada Nomor Kontrak: " + HelpersAM.parsingContractNo(json_data);
                //String errorMessage = e.getMessage();
                //log.error(errorMessage, e);
                //TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                e.printStackTrace();
                return String.valueOf(e);
            }
            //return "Success";
           // return null;
//        } catch (Exception e) {
////            log.error("Error Proses Bridging Pada Bagian: " + e.getMessage() + " Pada Nomor Kontrak: " + HelpersAM.parsingContractNo(json_data), e);
////            return "Error: " + e.getMessage() + " on Contract Number: " + HelpersAM.parsingContractNo(json_data);
//            //String errorMessage = e.getMessage();
//            //log.error(errorMessage, e);
//            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
//            return e.getMessage();
//        }
    }
}