@Slf4j
@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
public class BatchConfig {

  @Value("${custom.csv.tags.file-name}")
  private String fileNameTags;

  @Value("${custom.csv.authors.file-name}")
  private String fileNameAuthors;

  @Value("${custom.csv.books.file-name}")
  private String fileNameBooks;

  private final JobBuilderFactory jobBuilderFactory;
  private final StepBuilderFactory stepBuilderFactory;
  private final DataSource dataSource;

  @Bean
  public TagBatchImpl getTagData() {
    return new TagBatchImpl(
            dataSource,
            fileNameTags
    );
  }

  @Bean
  public AuthorsBatchImpl getAuthorData() {
    return new AuthorsBatchImpl(
            dataSource,
            fileNameAuthors
    );
  }

  @Bean
  public BooksBatchImpl getBookData() {
    return new BooksBatchImpl(
            dataSource,
            fileNameBooks
    );
  }

  @Bean
  public Step tagStep() {
    return stepBuilderFactory.get("tag-step")
            .<TagDto, TagDto>chunk(10)
            .reader(getTagData().getReader())
            .writer(getTagData().getWriter())
            .build();
  }

  @Bean
  public Step authorStep() {
    return stepBuilderFactory.get("author-step")
            .<AuthorDto, AuthorDto>chunk(10)
            .reader(getAuthorData().getReader())
            .writer(getAuthorData().getWriter())
            .build();
  }

  @Bean
  public Step bookStep() {
    return stepBuilderFactory.get("book-step")
            .<BookDto, BookDto>chunk(10)
            .reader(getBookData().getReader())
            .writer(getBookData().getWriter())
            .build();
  }

  @Bean
  public Job job() {
    return jobBuilderFactory.get("job")
            .start(tagStep())
            .next(authorStep())
            .next(bookStep())
            .build();
  }

}

public class AbstractBatchImpl<T> {

  private final DataSource dataSource;
  private final String fileName;
  private final Class<T> type;

  @Setter
  private String query;
  @Setter
  private String[] headers;

  public AbstractBatchImpl(
          DataSource dataSource,
          String fileName,
          Class<T> type) {
    this.dataSource = dataSource;
    this.fileName = fileName;
    this.type = type;
  }

  public FlatFileItemReader<T> getReader() {
    FlatFileItemReader<T> reader = new FlatFileItemReader<>();
    reader.setLinesToSkip(1);
    reader.setResource(new FileSystemResource(fileName));

    DefaultLineMapper<T> customerLineMapper = new DefaultLineMapper<>();

    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
    tokenizer.setNames(headers);

    customerLineMapper.setLineTokenizer(tokenizer);
    customerLineMapper.setFieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
      setTargetType(type);
    }});
    customerLineMapper.afterPropertiesSet();
    reader.setLineMapper(customerLineMapper);
    return reader;
  }

  @SuppressWarnings({ "rawtypes", "unchecked" })
  public JdbcBatchItemWriter<T> getWriter() {
    JdbcBatchItemWriter<T> itemWriter = new JdbcBatchItemWriter<>();

    itemWriter.setDataSource(dataSource);
    itemWriter.setSql(query);
    itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider());
    itemWriter.afterPropertiesSet();

    return itemWriter;
  }
}

@Slf4j
public class TagBatchImpl extends AbstractBatchImpl<TagDto> {

  public TagBatchImpl(
          DataSource dataSource,
          String fileName
  ) {
    super(dataSource, fileName, TagDto.class);
    setHeaders(new String[]{"id", "name"});
    setQuery(
            "SET IDENTITY_INSERT tags ON; " +
            "INSERT INTO tags (id, name) VALUES (:id, :name);"+
            "SET IDENTITY_INSERT tags OFF;"
    );
  }

}