/*
    * 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.shardingsphere.sharding.rule.checker;
  
  import com.google.common.base.Splitter;
  import lombok.RequiredArgsConstructor;
  import org.apache.shardingsphere.infra.algorithm.core.exception.AlgorithmInitializationException;
  import org.apache.shardingsphere.infra.datanode.DataNode;
  import org.apache.shardingsphere.infra.datanode.DataNodeInfo;
  import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
  import org.apache.shardingsphere.sharding.algorithm.sharding.inline.InlineShardingAlgorithm;
  import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
  import org.apache.shardingsphere.sharding.api.config.rule.ShardingTableReferenceRuleConfiguration;
  import org.apache.shardingsphere.sharding.api.config.strategy.sharding.ComplexShardingStrategyConfiguration;
  import org.apache.shardingsphere.sharding.api.config.strategy.sharding.ShardingStrategyConfiguration;
  import org.apache.shardingsphere.sharding.api.config.strategy.sharding.StandardShardingStrategyConfiguration;
  import org.apache.shardingsphere.sharding.api.sharding.standard.PreciseShardingValue;
  import org.apache.shardingsphere.sharding.exception.metadata.DuplicateSharingActualDataNodeException;
  import org.apache.shardingsphere.sharding.exception.metadata.InvalidBindingTablesException;
  import org.apache.shardingsphere.sharding.exception.metadata.ShardingTableRuleNotFoundException;
  import org.apache.shardingsphere.sharding.rule.BindingTableCheckedConfiguration;
  import org.apache.shardingsphere.sharding.rule.ShardingRule;
  import org.apache.shardingsphere.sharding.rule.ShardingTable;
  import org.apache.shardingsphere.sharding.spi.ShardingAlgorithm;
  
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.stream.Collectors;
  
  @RequiredArgsConstructor
  public class ShardingRuleChecker {
      
      private final ShardingRule shardingRule;
      
      /**
       * Check sharding rule.
       *
       * @param ruleConfig sharding rule configuration
       */
      public void check(final ShardingRuleConfiguration ruleConfig) {
          checkUniqueActualDataNodesInTableRules();
          checkBindingTableConfiguration(ruleConfig);
          checkInlineShardingAlgorithmsInTableRules();
      }
      
      private void checkUniqueActualDataNodesInTableRules() {
          Set<DataNode> uniqueActualDataNodes = new HashSet<>(shardingRule.getShardingTables().size(), 1L);
          shardingRule.getShardingTables().forEach((key, value) -> {
              DataNode sampleActualDataNode = value.getActualDataNodes().iterator().next();
              ShardingSpherePreconditions.checkState(!uniqueActualDataNodes.contains(sampleActualDataNode),
                      () -> new DuplicateSharingActualDataNodeException(key, sampleActualDataNode.getDataSourceName(), sampleActualDataNode.getTableName()));
              uniqueActualDataNodes.add(sampleActualDataNode);
          });
      }
      
      private void checkBindingTableConfiguration(final ShardingRuleConfiguration ruleConfig) {
          ShardingSpherePreconditions.checkState(
                  isValidBindingTableConfiguration(shardingRule.getShardingTables(), new BindingTableCheckedConfiguration(shardingRule.getDataSourceNames(), shardingRule.getShardingAlgorithms(),
                          ruleConfig.getBindingTableGroups(), shardingRule.getDefaultDatabaseShardingStrategyConfig(), shardingRule.getDefaultTableShardingStrategyConfig(),
                          shardingRule.getDefaultShardingColumn())),
                  InvalidBindingTablesException::new);
      }
      
      private boolean isValidBindingTableConfiguration(final Map<String, ShardingTable> shardingTables, final BindingTableCheckedConfiguration checkedConfig) {
          for (ShardingTableReferenceRuleConfiguration each : checkedConfig.getBindingTableGroups()) {
              Collection<String> bindingTables = Splitter.on(",").trimResults().splitToList(each.getReference());
              if (bindingTables.size() <= 1) {
                  continue;
              }
              Iterator<String> iterator = bindingTables.iterator();
              ShardingTable sampleShardingTable = getShardingTable(iterator.next(), shardingTables);
              while (iterator.hasNext()) {
                  ShardingTable shardingTable = getShardingTable(iterator.next(), shardingTables);
                  if (!isValidActualDataSourceName(sampleShardingTable, shardingTable) || !isValidActualTableName(sampleShardingTable, shardingTable)) {
                      return false;
                  }
                  if (!isBindingShardingAlgorithm(sampleShardingTable, shardingTable, true, checkedConfig) || !isBindingShardingAlgorithm(sampleShardingTable, shardingTable, false, checkedConfig)) {
                      return false;
                  }
             }
         }
         return true;
     }
     
     private ShardingTable getShardingTable(final String logicTableName, final Map<String, ShardingTable> shardingTables) {
         ShardingTable result = shardingTables.get(logicTableName);
         if (null != result) {
             return result;
         }
         throw new ShardingTableRuleNotFoundException(Collections.singleton(logicTableName));
     }
     
     private boolean isValidActualDataSourceName(final ShardingTable sampleShardingTable, final ShardingTable shardingTable) {
         return sampleShardingTable.getActualDataSourceNames().equals(shardingTable.getActualDataSourceNames());
     }
     
     private boolean isValidActualTableName(final ShardingTable sampleShardingTable, final ShardingTable shardingTable) {
         for (String each : sampleShardingTable.getActualDataSourceNames()) {
             Collection<String> sampleActualTableNames =
                     sampleShardingTable.getActualTableNames(each).stream().map(actualTableName -> actualTableName.replace(sampleShardingTable.getTableDataNode().getPrefix(), ""))
                             .collect(Collectors.toSet());
             Collection<String> actualTableNames =
                     shardingTable.getActualTableNames(each).stream().map(optional -> optional.replace(shardingTable.getTableDataNode().getPrefix(), "")).collect(Collectors.toSet());
             if (!sampleActualTableNames.equals(actualTableNames)) {
                 return false;
             }
         }
         return true;
     }
     
     private boolean isBindingShardingAlgorithm(final ShardingTable sampleShardingTable, final ShardingTable shardingTable, final boolean databaseAlgorithm,
                                                final BindingTableCheckedConfiguration checkedConfig) {
         return getAlgorithmExpression(sampleShardingTable, databaseAlgorithm, checkedConfig).equals(getAlgorithmExpression(shardingTable, databaseAlgorithm, checkedConfig));
     }
     
     private Optional<String> getAlgorithmExpression(final ShardingTable shardingTable, final boolean databaseAlgorithm, final BindingTableCheckedConfiguration checkedConfig) {
         ShardingStrategyConfiguration shardingStrategyConfig = databaseAlgorithm
                 ? shardingRule.getDatabaseShardingStrategyConfiguration(shardingTable)
                 : shardingRule.getTableShardingStrategyConfiguration(shardingTable);
         ShardingAlgorithm shardingAlgorithm = checkedConfig.getShardingAlgorithms().get(shardingStrategyConfig.getShardingAlgorithmName());
         String dataNodePrefix = databaseAlgorithm ? shardingTable.getDataSourceDataNode().getPrefix() : shardingTable.getTableDataNode().getPrefix();
         String shardingColumn = getShardingColumn(shardingStrategyConfig, shardingRule.getDefaultShardingColumn());
         return null == shardingAlgorithm ? Optional.empty() : shardingAlgorithm.getAlgorithmStructure(dataNodePrefix, shardingColumn);
     }
     
     private String getShardingColumn(final ShardingStrategyConfiguration shardingStrategyConfig, final String defaultShardingColumn) {
         String shardingColumn = defaultShardingColumn;
         if (shardingStrategyConfig instanceof ComplexShardingStrategyConfiguration) {
             shardingColumn = ((ComplexShardingStrategyConfiguration) shardingStrategyConfig).getShardingColumns();
         }
         if (shardingStrategyConfig instanceof StandardShardingStrategyConfiguration) {
             shardingColumn = ((StandardShardingStrategyConfiguration) shardingStrategyConfig).getShardingColumn();
         }
         return null == shardingColumn ? "" : shardingColumn;
     }
     
     private void checkInlineShardingAlgorithmsInTableRules() {
         shardingRule.getShardingTables().forEach((key, value) -> {
             validateInlineShardingAlgorithm(value, shardingRule.getTableShardingStrategyConfiguration(value), value.getTableDataNode());
             validateInlineShardingAlgorithm(value, shardingRule.getDatabaseShardingStrategyConfiguration(value), value.getDataSourceDataNode());
         });
     }
     
     private void validateInlineShardingAlgorithm(final ShardingTable shardingTable, final ShardingStrategyConfiguration shardingStrategy, final DataNodeInfo dataNodeInfo) {
         if (null == shardingStrategy) {
             return;
         }
         ShardingAlgorithm shardingAlgorithm = shardingRule.getShardingAlgorithms().get(shardingStrategy.getShardingAlgorithmName());
         if (shardingAlgorithm instanceof InlineShardingAlgorithm) {
             String shardingColumn = null == ((StandardShardingStrategyConfiguration) shardingStrategy).getShardingColumn() ? shardingRule.getDefaultShardingColumn()
                     : ((StandardShardingStrategyConfiguration) shardingStrategy).getShardingColumn();
             String result = null;
             try {
                 result = ((InlineShardingAlgorithm) shardingAlgorithm).doSharding(Collections.emptySet(), new PreciseShardingValue<>(shardingTable.getLogicTable(), shardingColumn, dataNodeInfo, 1));
                 // CHECKSTYLE:OFF
             } catch (final Exception ignored) {
                 // CHECKSTYLE:ON
             }
             ShardingSpherePreconditions.checkState(null == result || result.startsWith(dataNodeInfo.getPrefix()),
                     () -> new AlgorithmInitializationException(shardingAlgorithm, "`%s` sharding algorithm configuration of `%s` does not match the actual data nodes",
                             shardingStrategy.getShardingAlgorithmName(), shardingTable.getLogicTable()));
         }
     }
 }