/*
 * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 */

package io.airbyte.integrations.destination.s3.writer;

import com.amazonaws.services.s3.AmazonS3;
import io.airbyte.integrations.destination.s3.S3DestinationConfig;
import io.airbyte.integrations.destination.s3.S3Format;
import io.airbyte.integrations.destination.s3.avro.AvroConstants;
import io.airbyte.integrations.destination.s3.avro.JsonToAvroSchemaConverter;
import io.airbyte.integrations.destination.s3.avro.S3AvroWriter;
import io.airbyte.integrations.destination.s3.csv.S3CsvWriter;
import io.airbyte.integrations.destination.s3.jsonl.S3JsonlWriter;
import io.airbyte.integrations.destination.s3.parquet.S3ParquetWriter;
import io.airbyte.protocol.models.AirbyteStream;
import io.airbyte.protocol.models.ConfiguredAirbyteStream;
import java.sql.Timestamp;
import org.apache.avro.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProductionWriterFactory implements S3WriterFactory {

  protected static final Logger LOGGER = LoggerFactory.getLogger(ProductionWriterFactory.class);

  @Override
  public DestinationFileWriter create(final S3DestinationConfig config,
                                      final AmazonS3 s3Client,
                                      final ConfiguredAirbyteStream configuredStream,
                                      final Timestamp uploadTimestamp)
      throws Exception {
    final S3Format format = config.getFormatConfig().getFormat();

    if (format == S3Format.AVRO || format == S3Format.PARQUET) {
      final AirbyteStream stream = configuredStream.getStream();
      LOGGER.info("Json schema for stream {}: {}", stream.getName(), stream.getJsonSchema());

      final JsonToAvroSchemaConverter schemaConverter = new JsonToAvroSchemaConverter();
      final Schema avroSchema = schemaConverter.getAvroSchema(stream.getJsonSchema(), stream.getName(), stream.getNamespace());

      LOGGER.info("Avro schema for stream {}: {}", stream.getName(), avroSchema.toString(false));

      if (format == S3Format.AVRO) {
        return new S3AvroWriter(config, s3Client, configuredStream, uploadTimestamp, avroSchema, AvroConstants.JSON_CONVERTER);
      } else {
        return new S3ParquetWriter(config, s3Client, configuredStream, uploadTimestamp, avroSchema, AvroConstants.JSON_CONVERTER);
      }
    }

    if (format == S3Format.CSV) {
      return new S3CsvWriter.Builder(config, s3Client, configuredStream, uploadTimestamp).build();
    }

    if (format == S3Format.JSONL) {
      return new S3JsonlWriter(config, s3Client, configuredStream, uploadTimestamp);
    }

    throw new RuntimeException("Unexpected S3 destination format: " + format);
  }

}
