content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.camel.quarkus.component.aws2.kinesis.it;
  
  import java.util.List;
  import java.util.Locale;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.camel.quarkus.test.support.aws2.Aws2TestEnvContext;
  import org.apache.camel.quarkus.test.support.aws2.Aws2TestEnvCustomizer;
  import org.apache.commons.lang3.RandomStringUtils;
  import org.awaitility.Awaitility;
  import org.jboss.logging.Logger;
  import org.testcontainers.containers.localstack.LocalStackContainer.Service;
  import software.amazon.awssdk.services.firehose.FirehoseClient;
  import software.amazon.awssdk.services.firehose.model.BufferingHints;
  import software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamRequest;
  import software.amazon.awssdk.services.firehose.model.DeleteDeliveryStreamRequest;
  import software.amazon.awssdk.services.firehose.model.DeliveryStreamStatus;
  import software.amazon.awssdk.services.firehose.model.DeliveryStreamType;
  import software.amazon.awssdk.services.firehose.model.DescribeDeliveryStreamRequest;
  import software.amazon.awssdk.services.firehose.model.ExtendedS3DestinationConfiguration;
  import software.amazon.awssdk.services.firehose.model.InvalidArgumentException;
  import software.amazon.awssdk.services.iam.IamClient;
  import software.amazon.awssdk.services.iam.model.AttachRolePolicyRequest;
  import software.amazon.awssdk.services.iam.model.CreatePolicyRequest;
  import software.amazon.awssdk.services.iam.model.CreateRoleRequest;
  import software.amazon.awssdk.services.iam.model.DeletePolicyRequest;
  import software.amazon.awssdk.services.iam.model.DeleteRoleRequest;
  import software.amazon.awssdk.services.iam.model.DetachRolePolicyRequest;
  import software.amazon.awssdk.services.iam.model.GetPolicyRequest;
  import software.amazon.awssdk.services.iam.model.GetRoleRequest;
  import software.amazon.awssdk.services.iam.waiters.IamWaiter;
  import software.amazon.awssdk.services.kinesis.KinesisClient;
  import software.amazon.awssdk.services.kinesis.model.CreateStreamRequest;
  import software.amazon.awssdk.services.kinesis.model.DeleteStreamRequest;
  import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
  import software.amazon.awssdk.services.kinesis.waiters.KinesisWaiter;
  import software.amazon.awssdk.services.s3.S3Client;
  import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
  import software.amazon.awssdk.services.s3.model.DeleteBucketRequest;
  import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
  import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
  import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
  import software.amazon.awssdk.services.s3.model.ListObjectsResponse;
  import software.amazon.awssdk.services.s3.model.S3Object;
  import software.amazon.awssdk.services.s3.waiters.S3Waiter;
  
  public class Aws2KinesisTestEnvCustomizer implements Aws2TestEnvCustomizer {
      public static final int BUFFERING_SIZE_MB = 1;
      public static final int BUFFERING_TIME_SEC = 60;
      private static final Logger LOG = Logger.getLogger(Aws2KinesisTestEnvCustomizer.class);
  
      @Override
      public Service[] localstackServices() {
          return new Service[] { Service.KINESIS, Service.FIREHOSE, Service.S3, Service.IAM, Service.STS };
      }
  
      @Override
      public Service[] exportCredentialsForLocalstackServices() {
          return new Service[] { Service.KINESIS, Service.FIREHOSE };
      }
  
      @Override
      public void customize(Aws2TestEnvContext envContext) {
  
          final String streamName = "camel-quarkus-" + RandomStringUtils.secure().nextAlphanumeric(16).toLowerCase(Locale.ROOT);
          final String streamNameForDefaultCredentials = "camel-quarkus-"
                  + RandomStringUtils.secure().nextAlphanumeric(16).toLowerCase(Locale.ROOT);
          final String streamArn;
          {
              envContext.property("aws-kinesis.stream-name", streamName);
              final KinesisClient client = envContext.client(Service.KINESIS, KinesisClient::builder);
              client.createStream(
                      CreateStreamRequest.builder()
                              .shardCount(1)
                              .streamName(streamName)
                              .build());
  
              try (KinesisWaiter waiter = client.waiter()) {
                  streamArn = waiter.waitUntilStreamExists(DescribeStreamRequest.builder()
                          .streamName(streamName)
                          .build())
                          .matched().response().get().streamDescription().streamARN();
              }
  
              envContext.closeable(() -> client.deleteStream(DeleteStreamRequest.builder().streamName(streamName).build()));
          }
  
          {
              envContext.property("aws-kinesis.stream-name-for-default-credentials", streamNameForDefaultCredentials);
              final KinesisClient client = envContext.client(Service.KINESIS, KinesisClient::builder);
              client.createStream(
                      CreateStreamRequest.builder()
                              .shardCount(1)
                              .streamName(streamNameForDefaultCredentials)
                              .build());
  
              envContext.closeable(() -> client
                      .deleteStream(DeleteStreamRequest.builder().streamName(streamNameForDefaultCredentials).build()));
          }
  
          {
              final S3Client s3Client = envContext.client(Service.S3, S3Client::builder);
  
              final String bucketName = "camel-quarkus-firehose-"
                      + RandomStringUtils.secure().nextAlphanumeric(32).toLowerCase(Locale.ROOT);
              final String bucketArn = "arn:aws:s3:::" + bucketName;
              envContext.property("aws-kinesis.s3-bucket-name", bucketName);
              s3Client.createBucket(CreateBucketRequest.builder().bucket(bucketName).build());
              envContext.closeable(() -> {
                  LOG.info("Deleting bucket " + bucketName);
                  s3Client.deleteBucket(DeleteBucketRequest.builder().bucket(bucketName).build());
              });
              envContext.closeable(() -> {
                  LOG.info("Deleting objects from bucket bucket " + bucketName);
                  final ListObjectsResponse objects = s3Client.listObjects(
                          ListObjectsRequest.builder()
                                  .bucket(bucketName)
                                  .build());
                  final List<S3Object> objs = objects.contents();
                  LOG.info("Deleting " + objs.size() + " objects in bucket " + bucketName);
                  for (S3Object obj : objs) {
                      LOG.info("Deleting object " + obj.key());
                      s3Client.deleteObject(DeleteObjectRequest.builder().bucket(bucketName).key(obj.key()).build());
                  }
              });
  
              try (S3Waiter w = s3Client.waiter()) {
                  w.waitUntilBucketExists(HeadBucketRequest.builder().bucket(bucketName).build());
              }
  
              final String deliveryStreamName = "camel-quarkus-firehose-delstr-"
                      + RandomStringUtils.secure().nextAlphanumeric(16).toLowerCase(Locale.ROOT);
              envContext.property("aws-kinesis-firehose.delivery-stream-name", deliveryStreamName);
  
              final String roleName = "s3-" + deliveryStreamName;
  
              final IamClient iamClient = envContext.client(Service.IAM, IamClient::builder);
              final String roleArn = iamClient.createRole(
                      CreateRoleRequest.builder()
                              .roleName(roleName)
                              .path("/service-role/")
                              .assumeRolePolicyDocument("{\n"
                                      + "  \"Version\": \"2012-10-17\",\n"
                                      + "  \"Statement\": [\n"
                                      + "    {\n"
                                      + "      \"Sid\": \"sid" + RandomStringUtils.secure().nextAlphanumeric(16) + "\",\n"
                                      + "      \"Effect\": \"Allow\",\n"
                                      + "      \"Principal\": {\n"
                                      + "        \"Service\": \"firehose.amazonaws.com\"\n"
                                      + "      },\n"
                                      + "      \"Action\": \"sts:AssumeRole\"\n"
                                      + "    }\n"
                                      + "  ]\n"
                                      + "}")
                              .build())
                      .role().arn();
              envContext.closeable(() -> iamClient.deleteRole(DeleteRoleRequest.builder().roleName(roleName).build()));
  
              try (IamWaiter w = iamClient.waiter()) {
                  w.waitUntilRoleExists(GetRoleRequest.builder().roleName(roleName).build());
              }
  
              final String policyName = "firehose-s3-policy-" + deliveryStreamName;
  
              final String policy = "{\n"
                      + "    \"Version\": \"2012-10-17\",\n"
                      + "    \"Statement\":\n"
                      + "    [\n"
                      + "        {\n"
                      + "            \"Sid\": \"sid" + RandomStringUtils.secure().nextAlphanumeric(16) + "\",\n"
                      + "            \"Effect\": \"Allow\",\n"
                      + "            \"Action\": [\n"
                      + "                \"s3:AbortMultipartUpload\",\n"
                      + "                \"s3:GetBucketLocation\",\n"
                      + "                \"s3:GetObject\",\n"
                      + "                \"s3:ListBucket\",\n"
                      + "                \"s3:ListBucketMultipartUploads\",\n"
                      + "                \"s3:PutObject\"\n"
                      + "            ],      \n"
                      + "            \"Resource\": [\n"
                      + "                \"arn:aws:s3:::" + bucketName + "\",\n"
                      + "                \"arn:aws:s3:::" + bucketName + "/*\"\n"
                      + "            ]\n"
                      + "        },\n"
                      + "        {\n"
                      + "            \"Sid\": \"sid" + RandomStringUtils.secure().nextAlphanumeric(16) + "\",\n"
                      + "            \"Effect\": \"Allow\",\n"
                      + "            \"Action\": [\n"
                      + "                \"kinesis:DescribeStream\",\n"
                      + "                \"kinesis:GetShardIterator\",\n"
                      + "                \"kinesis:GetRecords\",\n"
                      + "                \"kinesis:ListShards\"\n"
                      + "            ],\n"
                      + "            \"Resource\": \"" + streamArn + "\"\n"
                      + "        }\n"
                      + "    ]\n"
                      + "}";
              final String policyArn = iamClient.createPolicy(
                      CreatePolicyRequest.builder()
                              .policyName(policyName)
                              .policyDocument(policy)
                              .build())
                      .policy().arn();
              envContext.closeable(() -> iamClient.deletePolicy(DeletePolicyRequest.builder().policyArn(policyArn).build()));
  
              try (IamWaiter w = iamClient.waiter()) {
                  w.waitUntilPolicyExists(GetPolicyRequest.builder().policyArn(policyArn).build());
              }
  
              iamClient.attachRolePolicy(
                      AttachRolePolicyRequest.builder()
                              .policyArn(policyArn)
                              .roleName(roleName)
                              .build());
              envContext.closeable(() -> iamClient.detachRolePolicy(
                      DetachRolePolicyRequest.builder()
                              .roleName(roleName)
                              .policyArn(policyArn)
                              .build()));
  
              final FirehoseClient fhClient = envContext.client(Service.FIREHOSE, FirehoseClient::builder);
  
              /*
               * Some of the dependency resources above needs some time to get visible for the firehose service
               * So we need to retry creation of the delivery stream until it succeeds
               */
              Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
                      () -> {
                          try {
                              fhClient.createDeliveryStream(
                                      CreateDeliveryStreamRequest.builder()
                                              .deliveryStreamName(deliveryStreamName)
                                              .extendedS3DestinationConfiguration(
                                                      ExtendedS3DestinationConfiguration.builder()
                                                              .bucketARN(bucketArn)
                                                              .roleARN(roleArn)
                                                              .bufferingHints(
                                                                      BufferingHints.builder()
                                                                              .intervalInSeconds(BUFFERING_TIME_SEC)
                                                                              .sizeInMBs(BUFFERING_SIZE_MB)
                                                                              .build())
                                                              .build())
                                              .deliveryStreamType(DeliveryStreamType.DIRECT_PUT)
                                              .build());
                              LOG.info("Firehose delivery stream " + deliveryStreamName + " finally created");
                              return true;
                          } catch (InvalidArgumentException e) {
                              LOG.info("Retrying the creation of delivery stream " + deliveryStreamName + " because "
                                      + e.getMessage());
                              return false;
                          }
                      });
  
              /*
               * There is no waiter for FirehoseClient so we are polling the state of the stream until the state is ACTIVE
               * Feel free to improve if you see a more elegant way to do this
               */
              Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
                      () -> {
                          DeliveryStreamStatus status = fhClient.describeDeliveryStream(
                                  DescribeDeliveryStreamRequest.builder()
                                          .deliveryStreamName(deliveryStreamName)
                                          .build())
                                  .deliveryStreamDescription().deliveryStreamStatus();
                          LOG.info("Delivery stream " + deliveryStreamName + " status: " + status);
                          return status == DeliveryStreamStatus.ACTIVE;
                      });
  
              envContext.closeable(() -> fhClient.deleteDeliveryStream(
                      DeleteDeliveryStreamRequest.builder().deliveryStreamName(deliveryStreamName).build()));
  
          }
  
      }
  }
  