package com.atlassian.messagequeue.internal.sqs;
   
   import com.amazonaws.services.sqs.AmazonSQS;
   import com.amazonaws.services.sqs.model.InvalidIdFormatException;
   import com.amazonaws.services.sqs.model.MessageNotInflightException;
   import com.amazonaws.services.sqs.model.ReceiptHandleIsInvalidException;
   import com.atlassian.messagequeue.MessageAcknowledgementException;
   import com.atlassian.messagequeue.MessageRunnerServiceException;
   import com.atlassian.messagequeue.registry.MessageContext;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  import java.time.Duration;
  import java.util.Optional;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * SQS message context.
   */
  class SQSMessageContext implements MessageContext {
      static final long TWELVE_HOURS_AS_SECONDS = Duration.ofHours(12).getSeconds();
  
      private static final Logger log = LoggerFactory.getLogger(SQSMessageContext.class);
  
      private final AmazonSQS amazonSQSClient;
      private final String messageId;
      @Nullable
      private final String payload;
      private final String queueUrl;
      private final String receiptHandle;
      private final AtomicBoolean acknowledged = new AtomicBoolean(false);
      private final AtomicBoolean shuttingDown;
      private boolean autoAcknowledgeMessage = true;
      private final Future<?> visibilityTimeoutExtensionFuture;
  
      SQSMessageContext(@Nonnull String messageId, @Nullable String payload, AmazonSQS amazonSQSClient,
                               String queueUrl, String receiptHandle, AtomicBoolean shuttingDown, Future<?> visibilityTimeoutExtensionFuture) {
          this.messageId = requireNonNull(messageId, "messageId");
          this.payload = payload;
          this.amazonSQSClient = requireNonNull(amazonSQSClient, "amazonSQSClient");
          this.queueUrl = requireNonNull(queueUrl, "queueUrl");
          this.receiptHandle = requireNonNull(receiptHandle, "receiptHandle");
          this.shuttingDown = requireNonNull(shuttingDown, "shuttingDown");
          this.visibilityTimeoutExtensionFuture = requireNonNull(visibilityTimeoutExtensionFuture, "visibilityTimeoutExtensionFuture");
      }
  
      @Override
      public Optional<String> getMessageId() {
          return Optional.of(messageId);
      }
  
      @Override
      public Optional<String> getPayload() {
          return Optional.ofNullable(payload);
      }
  
      @Override
      public void acknowledge() {
          if (acknowledged.compareAndSet(false, true)) {
              try {
                  amazonSQSClient.deleteMessage(queueUrl, receiptHandle);
  
                  if (log.isInfoEnabled()) {
                      log.info("Message acknowledged (messageId: {}, receiptHandle: {})", messageId, receiptHandle);
                  }
              } catch (ReceiptHandleIsInvalidException | InvalidIdFormatException e) {
                  throw new MessageAcknowledgementException("Error acknowledging message: receipt handle invalid " + receiptHandle, e);
              }
  
              if (!visibilityTimeoutExtensionFuture.isDone()) {
                  final boolean cancelled = visibilityTimeoutExtensionFuture.cancel(true);
                  log.info("Cancelled extension of visibility timeout (in response to message acknowledgement). Cancellation status: {}", cancelled);
              }
          }
      }
  
      @Override
      public boolean isCancellationRequested() {
          return shuttingDown.get();
      }
  
      synchronized void changeMessageProcessingTimeout(long timeout, TimeUnit timeUnit) {
          final long timeoutAsSeconds = timeUnit.toSeconds(timeout);
          if (timeoutAsSeconds > TWELVE_HOURS_AS_SECONDS) {
              throw new IllegalArgumentException("Timeout too high. Ensure timeout is, in seconds, between 0 to 43200 (12 hours)");
          }
  
          try {
              amazonSQSClient.changeMessageVisibility(queueUrl, receiptHandle, (int) timeoutAsSeconds); // safe cast since legal values must be less than 43200
          } catch (MessageNotInflightException | ReceiptHandleIsInvalidException e) {
              throw new MessageRunnerServiceException("Error changing message processing timeout: ", e);
          }
      }
 
     @Override
     public boolean shouldAutoAcknowledgeMessage() {
         return autoAcknowledgeMessage;
     }
 
     @Override
     public void cancelAutoAcknowledgementOfMessage() {
         autoAcknowledgeMessage = false;
     }
 }