code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.workers.storage;
  
  import io.airbyte.config.storage.CloudStorageConfigs.MinioConfig;
  import io.airbyte.config.storage.CloudStorageConfigs.S3Config;
  import io.airbyte.config.storage.DefaultS3ClientFactory;
  import io.airbyte.config.storage.MinioS3ClientFactory;
  import io.airbyte.workers.general.DocumentStoreClient;
  import java.nio.charset.StandardCharsets;
  import java.nio.file.Path;
  import java.util.Optional;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import software.amazon.awssdk.core.ResponseBytes;
  import software.amazon.awssdk.core.sync.RequestBody;
  import software.amazon.awssdk.services.s3.S3Client;
  import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
  import software.amazon.awssdk.services.s3.model.GetObjectRequest;
  import software.amazon.awssdk.services.s3.model.GetObjectResponse;
  import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
  import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
  import software.amazon.awssdk.services.s3.model.PutObjectRequest;
  
  /**
   * Document store on top of the S3Client.
   */
  public class S3DocumentStoreClient implements DocumentStoreClient {
  
    private static final Logger LOGGER = LoggerFactory.getLogger(S3DocumentStoreClient.class);
  
    private final String bucketName;
    private final Path root;
    private final S3Client s3Client;
  
    public static S3DocumentStoreClient minio(final MinioConfig config, final Path root) {
      return new S3DocumentStoreClient(
          new MinioS3ClientFactory(config).get(),
          config.getBucketName(),
          root);
    }
  
    public static S3DocumentStoreClient s3(final S3Config config, final Path root) {
      return new S3DocumentStoreClient(
          new DefaultS3ClientFactory(config).get(),
          config.getBucketName(),
          root);
    }
  
    public S3DocumentStoreClient(final S3Client s3Client, final String bucketName, final Path root) {
      this.s3Client = s3Client;
      this.bucketName = bucketName;
      this.root = root;
    }
  
    String getKey(final String id) {
      return root + "/" + id;
    }
  
    @Override
    public void write(final String id, final String document) {
      final PutObjectRequest request = PutObjectRequest.builder()
          .bucket(bucketName)
          .key(getKey(id))
          .build();
  
      s3Client.putObject(request, RequestBody.fromString(document));
    }
  
    @Override
    public Optional<String> read(final String id) {
      try {
        final ResponseBytes<GetObjectResponse> objectAsBytes = s3Client.getObjectAsBytes(GetObjectRequest.builder()
            .bucket(bucketName)
            .key(getKey(id))
            .build());
        return Optional.of(objectAsBytes.asString(StandardCharsets.UTF_8));
      } catch (final NoSuchKeyException e) {
        LOGGER.debug("Could not find record with id {}", id);
        return Optional.empty();
      }
    }
  
    @Override
    public boolean delete(final String id) {
      boolean keyExists = true;
      try {
        s3Client.headObject(HeadObjectRequest.builder().bucket(bucketName).key(getKey(id)).build());
      } catch (final NoSuchKeyException e) {
        keyExists = false;
      }
  
      s3Client.deleteObject(DeleteObjectRequest.builder().bucket(bucketName).key(getKey(id)).build());
      return keyExists;
    }
  
  }
  