package com.mns.oms.batch.config;

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

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.data.MongoItemReader;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

import com.mns.oms.batch.domain.CarrierData;
import com.mns.oms.batch.listener.CarrierStepListener;
import com.mns.oms.batch.listener.JobStatusNotificationListener;
import com.mns.oms.batch.model.BeamDataDTO;
import com.mns.oms.batch.processor.BeamDataProcessor;
import com.mns.oms.batch.writer.KafkaBatchWriter;
import org.springframework.transaction.PlatformTransactionManager;

/**
 * @author Mrinmoy Mandal
 *
 *         Module: WISMR
 *
 *
 */

@Configuration
@EnableBatchProcessing
@EnableScheduling
@ConditionalOnProperty(value = "beam.batchjob.enabled", matchIfMissing = true, havingValue = "true")
public class BeamDataBatchConfiguration {

	@Autowired
	private JobStatusNotificationListener jobListener;

	@Value("${beam.data.write.chunk.size}")
	private String chunkSize;

	@Autowired
	@Qualifier("beamTaskExecutor")
	private TaskExecutor beamTaskExecutor;

	@Value("${beam.batchjob.step.partitioner.each.range}")
	private int range;

	@Autowired
	private MongoTemplate mongoTemplate;

//	@Autowired
	private JobRepository jobRepository;
	@Autowired
	private PlatformTransactionManager transactionManager;

	@Autowired
	private JobLauncher jobLauncher;

	@Scheduled(cron = "${beam.spring.batch.job.cron.expression}")
	public void ffiSchedule() {
		try {
			JobParameters jobParameters = new JobParametersBuilder().addDate("launchDate", new Date())
					.toJobParameters();
			jobLauncher.run(exportDataToBeam(), jobParameters);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Bean
	public JobRepository jobRepository() throws Exception {
		MongoJobRepositoryFactoryBean factory = new MongoJobRepositoryFactoryBean();
		factory.setMongoTemplate(mongoTemplate);
		factory.afterPropertiesSet();
		return factory.getObject();
	}
	@Bean
	@StepScope
	public MongoItemReader<CarrierData> mongoItemReader(@Value("#{stepExecutionContext['minValue']}") Long minValue,
														@Value("#{stepExecutionContext['maxValue']}") Long maxValue) {

		MongoItemReader<CarrierData> reader = new MongoItemReader<>();
		reader.setTemplate(mongoTemplate);
		Map<String, Direction> sortMap = new HashMap<>();
		sortMap.put("_id", Direction.DESC);
		reader.setSort(sortMap);
		reader.setTargetType(CarrierData.class);
		reader.setPageSize(range);
		reader.setQuery("{isProcessed: {$eq: false} }");
		return reader;

	}

	@Bean
	public BeamDataProcessor beamDataProcessor() {

		return new BeamDataProcessor();
	}

	@Autowired
	private KafkaBatchWriter kafkaItemWriter;

	@Bean
	public Job exportDataToBeam() throws Exception {

		return new JobBuilder("exportDataToBeam", jobRepository)
				.incrementer(new RunIdIncrementer())
				.listener(jobListener)
				.start(beamMasterStep())
				.build();
	}

	@Bean
	public Step beamMasterStep() throws Exception {

		return new StepBuilder("beamStep", jobRepository)
				.<CarrierData, BeamDataDTO>chunk(Integer.valueOf(chunkSize), transactionManager)
				.reader(mongoItemReader(null, null))
				.processor(beamDataProcessor())
				.writer(kafkaItemWriter)
				.taskExecutor(beamTaskExecutor)
				.listener(new CarrierStepListener())
				.build();

	}

}