package com.atlassian.messagequeue.internal.sqs.yaml;
   
   import com.atlassian.messagequeue.MessageRunnerKey;
   import com.atlassian.messagequeue.internal.sqs.DefaultSQSConsumerQueueConfig;
   import com.atlassian.messagequeue.internal.sqs.DefaultSQSProducerQueueConfig;
   import com.atlassian.messagequeue.internal.sqs.SQSConfig;
   import com.atlassian.messagequeue.internal.sqs.SQSConsumerQueueConfig;
   import com.atlassian.messagequeue.internal.sqs.SQSProducerQueueConfig;
   import org.yaml.snakeyaml.Yaml;
  
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Set;
  import java.util.function.Function;
  import java.util.function.Predicate;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  
  import static java.util.stream.Collectors.toSet;
  
  /**
   * Default Yaml config parser for sqs related config.
   */
  public class SQSYamlConfigParser {
  
      /**
       * Parse yaml config
       *
       * @param yamlConfigSupplier        content of Yaml config
       * @param queueNameValidator        queue name validator which validates every string queue name
       * @param queueNameToQueueUrlMapper queue name to queue url mapper.
       * @return {@link SQSConfig}
       */
      public SQSConfig parse(Supplier<String> yamlConfigSupplier, Predicate<String> queueNameValidator, Function<String, String> queueNameToQueueUrlMapper) {
          SQSProducerQueueConfig defaultSQSProducerQueueConfig = null;
          final Map<MessageRunnerKey, SQSProducerQueueConfig> messageRunnerKeyToQueueMappings = new HashMap<>();
          final Map<String, Set<SQSConsumerQueueConfig>> workerGroupToQueueMappings = new HashMap<>();
  
          final YamlConfig yamlConfig = new Yaml().loadAs(yamlConfigSupplier.get(), YamlConfig.class);
  
          validate(yamlConfig, queueNameValidator);
  
          for (YamlQueueConfig queue : yamlConfig.getQueues()) {
              final DefaultSQSProducerQueueConfig sqsProducerQueueConfig = new DefaultSQSProducerQueueConfig(queue.getQueueName(), queueNameToQueueUrlMapper.apply(queue.getQueueName()));
              if (queue.isDefault()) {
                  defaultSQSProducerQueueConfig = sqsProducerQueueConfig;
              }
  
              for (YamlMessageRunnerConfig messageRunner : queue.getMessageRunners()) {
                  messageRunnerKeyToQueueMappings.put(MessageRunnerKey.of(messageRunner.getKey()), sqsProducerQueueConfig);
              }
          }
  
          // build workerGroupToQueueMappings
          workerGroupToQueueMappings.clear();
          yamlConfig.getWorkerGroups().stream()
                  .forEach(workerGroup -> workerGroupToQueueMappings.put(workerGroup.getName(),
                          workerGroup.getConsumerQueues().stream()
                                  .map(config -> new DefaultSQSConsumerQueueConfig(config.getCorePoolSize(), config.getMaxPoolSize(),
                                          config.getVisibilityTimeoutExtensionPeriod(), config.getQueueName(), queueNameToQueueUrlMapper.apply(config.getQueueName())))
                                  .collect(toSet())));
  
          return new SQSConfig(defaultSQSProducerQueueConfig, messageRunnerKeyToQueueMappings, workerGroupToQueueMappings);
      }
  
      /**
       * Validate yaml config
       *
       * @param yamlConfig         loaded config
       * @param queueNameValidator validate queue name
       */
      private void validate(YamlConfig yamlConfig, Predicate<String> queueNameValidator) {
          if (yamlConfig.getQueues().size() == 0) {
              throw new RuntimeException("Invalid SQS config: there must be at least 1 queue");
          }
  
          for (YamlQueueConfig queue : yamlConfig.getQueues()) {
              if (!queueNameValidator.test(queue.getQueueName())) {
                  throw new RuntimeException("Invalid SQS queue name: " + queue.getQueueName());
              }
          }
  
          if (yamlConfig.getQueues().stream().filter(YamlQueueConfig::isDefault).count() != 1) {
              throw new RuntimeException("Invalid SQS config: there must be exactly 1 default queue");
          }
  
          if (yamlConfig.getWorkerGroups().size() == 0) {
              throw new RuntimeException("Invalid SQS config: there must be at least 1 worker group");
          }
  
          for (YamlWorkerGroupConfig workerGroupConfig : yamlConfig.getWorkerGroups()) {
              validateWorkerGroupConfig(workerGroupConfig);
          }
  
          final Set<String> allWorkerGroupQueues = yamlConfig.getWorkerGroups().stream()
                  .flatMap(workerGroup -> workerGroup.getConsumerQueues().stream())
                  .map(YamlConsumerQueueConfig::getQueueName)
                  .collect(toSet());
  
         final Set<YamlQueueConfig> queuesNotBelongToWorkerGroups = yamlConfig.getQueues().stream()
                 .filter(queue -> !allWorkerGroupQueues.contains(queue.getQueueName()))
                 .collect(toSet());
 
         if (queuesNotBelongToWorkerGroups.size() > 0) {
             final String queueNamesNotBelongToWorkerGroups = queuesNotBelongToWorkerGroups.stream()
                     .map(YamlQueueConfig::getQueueName)
                     .sorted()
                     .collect(Collectors.joining(", "));
 
             throw new RuntimeException("Invalid SQS config: queues(" + queueNamesNotBelongToWorkerGroups + ") must belong to at least 1 worker group");
         }
     }
 
     private void validateWorkerGroupConfig(YamlWorkerGroupConfig groupConfig) {
         for (YamlConsumerQueueConfig consumerQueueConfig : groupConfig.getConsumerQueues()) {
             if (consumerQueueConfig.getCorePoolSize() <= 0) {
                 throw new RuntimeException("Invalid SQS consumer queue config: corePoolSize must be greater than 0");
             }
 
             if (consumerQueueConfig.getMaxPoolSize() <= 0) {
                 throw new RuntimeException("Invalid SQS consumer queue config: maxPoolSize must be greater than 0");
             }
 
             if (consumerQueueConfig.getVisibilityTimeoutExtensionPeriod() <= 0) {
                 throw new RuntimeException("Invalid SQS consumer queue config: visibilityTimeoutExtensionPeriod must be greater than 0");
             }
         }
     }
 }