/*
 * 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 com.datastax.driver.core.schemabuilder;

import static java.util.Map.Entry;

import com.datastax.driver.core.DataType;
import com.datastax.driver.core.utils.MoreObjects;
import com.google.common.base.Joiner;
import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/** A built CREATE TABLE statement. */
public class Create extends AbstractCreateStatement<Create> {

  private String tableName;
  private Map<String, ColumnType> partitionColumns = new LinkedHashMap<String, ColumnType>();
  private Map<String, ColumnType> clusteringColumns = new LinkedHashMap<String, ColumnType>();
  private Map<String, ColumnType> staticColumns = new LinkedHashMap<String, ColumnType>();

  Create(String keyspaceName, String tableName) {
    validateNotEmpty(keyspaceName, "Keyspace name");
    validateNotEmpty(tableName, "Table name");
    validateNotKeyWord(
        keyspaceName,
        String.format(
            "The keyspace name '%s' is not allowed because it is a reserved keyword",
            keyspaceName));
    validateNotKeyWord(
        tableName,
        String.format(
            "The table name '%s' is not allowed because it is a reserved keyword", tableName));
    this.tableName = tableName;
    this.keyspaceName = Optional.fromNullable(keyspaceName);
  }

  Create(String tableName) {
    validateNotEmpty(tableName, "Table name");
    validateNotKeyWord(
        tableName,
        String.format(
            "The table name '%s' is not allowed because it is a reserved keyword", tableName));
    this.tableName = tableName;
  }

  /**
   * Add a partition key column definition to this CREATE TABLE statement.
   *
   * <p>This includes the column declaration (you don't need an additional {@code addColumn} call).
   *
   * <p>Partition key columns are added in the order of their declaration.
   *
   * @param columnName the name of the partition key column to be added.
   * @param dataType the data type of the partition key column to be added.
   * @return this CREATE statement.
   */
  public Create addPartitionKey(String columnName, DataType dataType) {
    validateNotEmpty(columnName, "Partition key name");
    validateNotNull(dataType, "Partition key type");
    validateNotKeyWord(
        columnName,
        String.format(
            "The partition key name '%s' is not allowed because it is a reserved keyword",
            columnName));
    partitionColumns.put(columnName, new NativeColumnType(dataType));
    return this;
  }

  /**
   * Add a partition key column definition to this CREATE TABLE statement, when its type contains a
   * UDT.
   *
   * <p>This includes the column declaration (you don't need an additional {@code addColumn} call).
   *
   * <p>Partition key columns are added in the order of their declaration.
   *
   * @param columnName the name of the partition key column to be added.
   * @param udtType the UDT type of the partition key column to be added. Use {@link
   *     SchemaBuilder#frozen(String)} or {@link SchemaBuilder#udtLiteral(String)}.
   * @return this CREATE statement.
   */
  public Create addUDTPartitionKey(String columnName, UDTType udtType) {
    validateNotEmpty(columnName, "Clustering key name");
    validateNotNull(udtType, "UDT partition key type");
    validateNotKeyWord(
        columnName,
        String.format(
            "The partition key name '%s' is not allowed because it is a reserved keyword",
            columnName));
    partitionColumns.put(columnName, udtType);
    return this;
  }

  /**
   * Add a clustering column definition to this CREATE TABLE statement.
   *
   * <p>This includes the column declaration (you don't need an additional {@code addColumn} call).
   *
   * <p>Clustering columns are added in the order of their declaration.
   *
   * @param columnName the name of the clustering column to be added.
   * @param dataType the data type of the clustering column to be added.
   * @return this CREATE statement.
   */
  public Create addClusteringColumn(String columnName, DataType dataType) {
    validateNotEmpty(columnName, "Clustering column name");
    validateNotNull(dataType, "Clustering column type");
    validateNotKeyWord(
        columnName,
        String.format(
            "The clustering column name '%s' is not allowed because it is a reserved keyword",
            columnName));
    clusteringColumns.put(columnName, new NativeColumnType(dataType));
    return this;
  }

  /**
   * Add a clustering column definition to this CREATE TABLE statement, when its type contains a
   * UDT.
   *
   * <p>This includes the column declaration (you don't need an additional {@code addColumn} call).
   *
   * <p>Clustering columns are added in the order of their declaration.
   *
   * @param columnName the name of the clustering column to be added.
   * @param udtType the UDT type of the clustering column to be added. Use {@link
   *     SchemaBuilder#frozen(String)} or {@link SchemaBuilder#udtLiteral(String)}.
   * @return this CREATE statement.
   */
  public Create addUDTClusteringColumn(String columnName, UDTType udtType) {
    validateNotEmpty(columnName, "Clustering column name");
    validateNotNull(udtType, "UDT clustering column type");
    validateNotKeyWord(
        columnName,
        String.format(
            "The clustering column name '%s' is not allowed because it is a reserved keyword",
            columnName));
    clusteringColumns.put(columnName, udtType);
    return this;
  }

  /**
   * Add a static column definition to this CREATE TABLE statement.
   *
   * @param columnName the name of the column to be added.
   * @param dataType the data type of the column to be added.
   * @return this CREATE statement.
   */
  public Create addStaticColumn(String columnName, DataType dataType) {
    validateNotEmpty(columnName, "Column name");
    validateNotNull(dataType, "Column type");
    validateNotKeyWord(
        columnName,
        String.format(
            "The static column name '%s' is not allowed because it is a reserved keyword",
            columnName));
    staticColumns.put(columnName, new NativeColumnType(dataType));
    return this;
  }

  /**
   * Add a static column definition to this CREATE TABLE statement, when its type contains a UDT.
   *
   * @param columnName the name of the column to be added.
   * @param udtType the UDT type of the column to be added. Use {@link SchemaBuilder#frozen(String)}
   *     or {@link SchemaBuilder#udtLiteral(String)}.
   * @return this CREATE statement.
   */
  public Create addUDTStaticColumn(String columnName, UDTType udtType) {
    validateNotEmpty(tableName, "Column name");
    validateNotNull(udtType, "Column UDT type");
    validateNotKeyWord(
        columnName,
        String.format(
            "The static column name '%s' is not allowed because it is a reserved keyword",
            columnName));
    staticColumns.put(columnName, udtType);
    return this;
  }

  /**
   * Add options for this CREATE TABLE statement.
   *
   * @return the options of this CREATE TABLE statement.
   */
  public Options withOptions() {
    return new Options(this);
  }

  /** The table options of a CREATE TABLE statement. */
  public static class Options extends TableOptions<Options> {

    private final Create create;

    private Options(Create create) {
      super(create.asStatementStart());
      this.create = create;
    }

    private List<ClusteringOrder> clusteringOrderKeys = Lists.newArrayList();

    private boolean compactStorage;

    /**
     * Add a clustering order for this table.
     *
     * <p>To define the order on multiple columns, call this method repeatedly. The columns will be
     * declared in the call order.
     *
     * @param columnName the clustering column name.
     * @param direction the clustering direction (DESC/ASC).
     * @return this {@code Options} object.
     */
    public Options clusteringOrder(String columnName, SchemaBuilder.Direction direction) {
      if (!create.clusteringColumns.containsKey(columnName)) {
        throw new IllegalArgumentException(
            String.format(
                "Clustering key '%s' is unknown. Did you forget to declare it first?", columnName));
      }
      clusteringOrderKeys.add(new ClusteringOrder(columnName, direction));
      return this;
    }

    /**
     * Enable the compact storage option for the table.
     *
     * @return this {@code Options} object.
     */
    public Options compactStorage() {
      this.compactStorage = true;
      return this;
    }

    private static class ClusteringOrder {
      private final String clusteringColumnName;
      private final SchemaBuilder.Direction direction;

      ClusteringOrder(String clusteringColumnName, SchemaBuilder.Direction direction) {
        validateNotEmpty(clusteringColumnName, "Column name for clustering order");
        this.clusteringColumnName = clusteringColumnName;
        this.direction = direction;
      }

      public String getClusteringColumnName() {
        return clusteringColumnName;
      }

      public String toStatement() {
        return clusteringColumnName + " " + direction.name();
      }

      @Override
      public String toString() {
        return toStatement();
      }

      @Override
      public boolean equals(Object o) {
        if (this == o) {
          return true;
        }
        if (o instanceof ClusteringOrder) {
          ClusteringOrder that = (ClusteringOrder) o;
          return MoreObjects.equal(this.clusteringColumnName, that.clusteringColumnName)
              && MoreObjects.equal(this.direction, that.direction);
        }
        return false;
      }

      @Override
      public int hashCode() {
        return MoreObjects.hashCode(clusteringColumnName, direction);
      }
    }

    @Override
    protected void addSpecificOptions(List<String> options) {
      if (!clusteringOrderKeys.isEmpty()) {
        options.add("CLUSTERING ORDER BY(" + Joiner.on(", ").join(clusteringOrderKeys) + ")");
      }

      if (compactStorage) {
        if (!create.staticColumns.isEmpty()) {
          throw new IllegalStateException(
              String.format(
                  "Cannot create table '%s' with compact storage and static columns '%s'",
                  create.tableName, create.staticColumns.keySet()));
        }
        options.add("COMPACT STORAGE");
      }
    }
  }

  @Override
  public String buildInternal() {
    if (partitionColumns.size() < 1) {
      throw new IllegalStateException(
          String.format(
              "There should be at least one partition key defined for the table '%s'", tableName));
    }

    validateColumnsDeclaration();

    StringBuilder createStatement = new StringBuilder(STATEMENT_START).append("CREATE TABLE");
    if (ifNotExists) {
      createStatement.append(" IF NOT EXISTS");
    }
    createStatement.append(" ");
    if (keyspaceName.isPresent()) {
      createStatement.append(keyspaceName.get()).append(".");
    }
    createStatement.append(tableName);

    List<String> allColumns = new ArrayList<String>();
    List<String> partitionKeyColumns = new ArrayList<String>();
    List<String> clusteringKeyColumns = new ArrayList<String>();

    for (Entry<String, ColumnType> entry : partitionColumns.entrySet()) {
      allColumns.add(entry.getKey() + " " + entry.getValue().asCQLString());
      partitionKeyColumns.add(entry.getKey());
    }

    for (Entry<String, ColumnType> entry : clusteringColumns.entrySet()) {
      allColumns.add(entry.getKey() + " " + entry.getValue().asCQLString());
      clusteringKeyColumns.add(entry.getKey());
    }

    for (Entry<String, ColumnType> entry : staticColumns.entrySet()) {
      allColumns.add(entry.getKey() + " " + entry.getValue().asCQLString() + " static");
    }

    for (Entry<String, ColumnType> entry : simpleColumns.entrySet()) {
      allColumns.add(buildColumnType(entry));
    }

    String partitionKeyPart =
        partitionKeyColumns.size() == 1
            ? partitionKeyColumns.get(0)
            : "(" + Joiner.on(", ").join(partitionKeyColumns) + ")";

    String primaryKeyPart =
        clusteringKeyColumns.size() == 0
            ? partitionKeyPart
            : partitionKeyPart + ", " + Joiner.on(", ").join(clusteringKeyColumns);

    createStatement.append("(").append(COLUMN_FORMATTING);
    createStatement.append(Joiner.on("," + COLUMN_FORMATTING).join(allColumns));
    createStatement.append("," + COLUMN_FORMATTING).append("PRIMARY KEY");
    createStatement.append("(").append(primaryKeyPart).append(")");
    createStatement.append(")");

    return createStatement.toString();
  }

  private void validateColumnsDeclaration() {

    final Collection partitionAndClusteringColumns =
        this.intersection(partitionColumns.keySet(), clusteringColumns.keySet());
    final Collection partitionAndSimpleColumns =
        this.intersection(partitionColumns.keySet(), simpleColumns.keySet());
    final Collection clusteringAndSimpleColumns =
        this.intersection(clusteringColumns.keySet(), simpleColumns.keySet());
    final Collection partitionAndStaticColumns =
        this.intersection(partitionColumns.keySet(), staticColumns.keySet());
    final Collection clusteringAndStaticColumns =
        this.intersection(clusteringColumns.keySet(), staticColumns.keySet());
    final Collection simpleAndStaticColumns =
        this.intersection(simpleColumns.keySet(), staticColumns.keySet());

    if (!partitionAndClusteringColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The '%s' columns can not be declared as partition keys and clustering keys at the same time",
              partitionAndClusteringColumns));
    }

    if (!partitionAndSimpleColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The '%s' columns can not be declared as partition keys and simple columns at the same time",
              partitionAndSimpleColumns));
    }

    if (!clusteringAndSimpleColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The '%s' columns can not be declared as clustering keys and simple columns at the same time",
              clusteringAndSimpleColumns));
    }

    if (!partitionAndStaticColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The '%s' columns can not be declared as partition keys and static columns at the same time",
              partitionAndStaticColumns));
    }

    if (!clusteringAndStaticColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The '%s' columns can not be declared as clustering keys and static columns at the same time",
              clusteringAndStaticColumns));
    }

    if (!simpleAndStaticColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The '%s' columns can not be declared as simple columns and static columns at the same time",
              simpleAndStaticColumns));
    }

    if (!staticColumns.isEmpty() && clusteringColumns.isEmpty()) {
      throw new IllegalStateException(
          String.format(
              "The table '%s' cannot declare static columns '%s' without clustering columns",
              tableName, staticColumns.keySet()));
    }
  }

  private <T> Collection<T> intersection(Collection<T> col1, Collection<T> col2) {
    Set<T> set = new HashSet<T>();

    for (T t : col1) {
      if (col2.contains(t)) {
        set.add(t);
      }
    }
    return set;
  }
}
